/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.ignite.internal.sql.engine.framework;

import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
import static org.apache.ignite.internal.sql.engine.exec.ExecutionServiceImplTest.PLANNING_THREAD_COUNT;
import static org.apache.ignite.internal.sql.engine.exec.ExecutionServiceImplTest.PLANNING_TIMEOUT;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.apache.ignite.internal.util.CollectionUtils.nullOrEmpty;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.mockito.Mockito.mock;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.BitSet;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.Flow.Publisher;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import org.apache.calcite.util.ImmutableIntList;
import org.apache.ignite.internal.catalog.CatalogCommand;
import org.apache.ignite.internal.catalog.CatalogManager;
import org.apache.ignite.internal.catalog.CatalogTestUtils;
import org.apache.ignite.internal.catalog.commands.CatalogUtils;
import org.apache.ignite.internal.catalog.commands.ColumnParams;
import org.apache.ignite.internal.catalog.commands.ColumnParams.Builder;
import org.apache.ignite.internal.catalog.commands.CreateHashIndexCommand;
import org.apache.ignite.internal.catalog.commands.CreateSortedIndexCommand;
import org.apache.ignite.internal.catalog.commands.CreateTableCommand;
import org.apache.ignite.internal.catalog.commands.DefaultValue;
import org.apache.ignite.internal.catalog.commands.MakeIndexAvailableCommand;
import org.apache.ignite.internal.catalog.commands.StartBuildingIndexCommand;
import org.apache.ignite.internal.catalog.commands.TableHashPrimaryKey;
import org.apache.ignite.internal.catalog.commands.TablePrimaryKey;
import org.apache.ignite.internal.catalog.descriptors.CatalogColumnCollation;
import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
import org.apache.ignite.internal.catalog.events.CatalogEvent;
import org.apache.ignite.internal.catalog.events.CreateIndexEventParameters;
import org.apache.ignite.internal.catalog.events.MakeIndexAvailableEventParameters;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalNode;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologySnapshot;
import org.apache.ignite.internal.event.EventListener;
import org.apache.ignite.internal.hlc.ClockWaiter;
import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.hlc.HybridClockImpl;
import org.apache.ignite.internal.hlc.TestClockService;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.metrics.MetricManagerImpl;
import org.apache.ignite.internal.sql.engine.SqlQueryProcessor;
import org.apache.ignite.internal.sql.engine.exec.ExecutableTable;
import org.apache.ignite.internal.sql.engine.exec.ExecutableTableRegistry;
import org.apache.ignite.internal.sql.engine.exec.ExecutionContext;
import org.apache.ignite.internal.sql.engine.exec.NodeWithConsistencyToken;
import org.apache.ignite.internal.sql.engine.exec.PartitionWithConsistencyToken;
import org.apache.ignite.internal.sql.engine.exec.QueryTaskExecutor;
import org.apache.ignite.internal.sql.engine.exec.RowHandler.RowFactory;
import org.apache.ignite.internal.sql.engine.exec.ScannableTable;
import org.apache.ignite.internal.sql.engine.exec.TxAttributes;
import org.apache.ignite.internal.sql.engine.exec.UpdatableTable;
import org.apache.ignite.internal.sql.engine.exec.ddl.DdlCommandHandler;
import org.apache.ignite.internal.sql.engine.exec.exp.RangeCondition;
import org.apache.ignite.internal.sql.engine.exec.mapping.ExecutionTarget;
import org.apache.ignite.internal.sql.engine.exec.mapping.ExecutionTargetFactory;
import org.apache.ignite.internal.sql.engine.exec.mapping.ExecutionTargetProvider;
import org.apache.ignite.internal.sql.engine.exec.mapping.FragmentDescription;
import org.apache.ignite.internal.sql.engine.exec.mapping.MappingServiceImpl;
import org.apache.ignite.internal.sql.engine.prepare.PrepareServiceImpl;
import org.apache.ignite.internal.sql.engine.prepare.ddl.DdlSqlToCommandConverter;
import org.apache.ignite.internal.sql.engine.prepare.pruning.PartitionPrunerImpl;
import org.apache.ignite.internal.sql.engine.schema.ColumnDescriptor;
import org.apache.ignite.internal.sql.engine.schema.ColumnDescriptorImpl;
import org.apache.ignite.internal.sql.engine.schema.DefaultValueStrategy;
import org.apache.ignite.internal.sql.engine.schema.IgniteIndex;
import org.apache.ignite.internal.sql.engine.schema.IgniteIndex.Collation;
import org.apache.ignite.internal.sql.engine.schema.IgniteSystemView;
import org.apache.ignite.internal.sql.engine.schema.IgniteTable;
import org.apache.ignite.internal.sql.engine.schema.IgniteTableImpl;
import org.apache.ignite.internal.sql.engine.schema.PartitionCalculator;
import org.apache.ignite.internal.sql.engine.schema.SqlSchemaManager;
import org.apache.ignite.internal.sql.engine.schema.SqlSchemaManagerImpl;
import org.apache.ignite.internal.sql.engine.schema.TableDescriptor;
import org.apache.ignite.internal.sql.engine.schema.TableDescriptorImpl;
import org.apache.ignite.internal.sql.engine.sql.ParserServiceImpl;
import org.apache.ignite.internal.sql.engine.trait.IgniteDistribution;
import org.apache.ignite.internal.sql.engine.trait.IgniteDistributions;
import org.apache.ignite.internal.sql.engine.util.Commons;
import org.apache.ignite.internal.sql.engine.util.EmptyCacheFactory;
import org.apache.ignite.internal.sql.engine.util.cache.CaffeineCacheFactory;
import org.apache.ignite.internal.systemview.SystemViewManagerImpl;
import org.apache.ignite.internal.systemview.api.SystemView;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.type.BitmaskNativeType;
import org.apache.ignite.internal.type.DecimalNativeType;
import org.apache.ignite.internal.type.NativeType;
import org.apache.ignite.internal.type.NativeTypeSpec;
import org.apache.ignite.internal.type.NumberNativeType;
import org.apache.ignite.internal.type.TemporalNativeType;
import org.apache.ignite.internal.type.VarlenNativeType;
import org.apache.ignite.internal.util.ArrayUtils;
import org.apache.ignite.internal.util.SubscriptionUtils;
import org.apache.ignite.internal.util.TransformingIterator;
import org.apache.ignite.internal.util.subscription.TransformingPublisher;
import org.apache.ignite.lang.ErrorGroups.Sql;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.sql.SqlException;
import org.jetbrains.annotations.Nullable;

/**
 * A collection of builders to create test objects.
 */
public class TestBuilders {
    private static final AtomicInteger TABLE_ID_GEN = new AtomicInteger();

    private static final IgniteLogger LOG = Loggers.forClass(TestBuilders.class);

    /** Returns a builder of the test cluster object. */
    public static ClusterBuilder cluster() {
        return new ClusterBuilderImpl();
    }

    /** Returns a builder of the test table object. */
    public static TableBuilder table() {
        return new TableBuilderImpl();
    }

    /** Returns a builder of the execution context. */
    public static ExecutionContextBuilder executionContext() {
        return new ExecutionContextBuilderImpl();
    }

    /** Factory method to create a cluster service factory for cluster consisting of provided nodes. */
    public static ClusterServiceFactory clusterServiceFactory(List<String> nodes) {
        return new ClusterServiceFactory(nodes);
    }

    /**
     * Factory method to create {@link ScannableTable table} instance from given data provider with
     * only implemented {@link ScannableTable#scan table scan}.
     */
    public static ScannableTable tableScan(DataProvider<Object[]> dataProvider) {
        return new ScannableTable() {
            @Override
            public <RowT> Publisher<RowT> scan(
                    ExecutionContext<RowT> ctx,
                    PartitionWithConsistencyToken partWithConsistencyToken,
                    RowFactory<RowT> rowFactory,
                    @Nullable BitSet requiredColumns
            ) {

                return new TransformingPublisher<>(
                        SubscriptionUtils.fromIterable(
                                () -> new TransformingIterator<>(
                                        dataProvider.iterator(),
                                        row -> project(row, requiredColumns)
                                )
                        ),
                        rowFactory::create
                );
            }

            @Override
            public <RowT> Publisher<RowT> indexRangeScan(ExecutionContext<RowT> ctx, PartitionWithConsistencyToken partWithConsistencyToken,
                    RowFactory<RowT> rowFactory, int indexId, List<String> columns, @Nullable RangeCondition<RowT> cond,
                    @Nullable BitSet requiredColumns) {
                throw new UnsupportedOperationException();
            }

            @Override
            public <RowT> Publisher<RowT> indexLookup(ExecutionContext<RowT> ctx, PartitionWithConsistencyToken partWithConsistencyToken,
                    RowFactory<RowT> rowFactory, int indexId, List<String> columns, RowT key, @Nullable BitSet requiredColumns) {
                throw new UnsupportedOperationException();
            }

            @Override
            public <RowT> CompletableFuture<@Nullable RowT> primaryKeyLookup(ExecutionContext<RowT> ctx, InternalTransaction tx,
                    RowFactory<RowT> rowFactory, RowT key, @Nullable BitSet requiredColumns) {
                throw new UnsupportedOperationException();
            }
        };
    }

    /**
     * Factory method to create {@link ScannableTable table} instance from given data provider with
     * only implemented {@link ScannableTable#indexRangeScan index range scan}.
     */
    public static ScannableTable indexRangeScan(DataProvider<Object[]> dataProvider) {
        return new ScannableTable() {
            @Override
            public <RowT> Publisher<RowT> scan(
                    ExecutionContext<RowT> ctx,
                    PartitionWithConsistencyToken partWithConsistencyToken,
                    RowFactory<RowT> rowFactory,
                    @Nullable BitSet requiredColumns
            ) {
                throw new UnsupportedOperationException();
            }

            @Override
            public <RowT> Publisher<RowT> indexRangeScan(ExecutionContext<RowT> ctx, PartitionWithConsistencyToken partWithConsistencyToken,
                    RowFactory<RowT> rowFactory, int indexId, List<String> columns, @Nullable RangeCondition<RowT> cond,
                    @Nullable BitSet requiredColumns) {
                return new TransformingPublisher<>(
                        SubscriptionUtils.fromIterable(
                                () -> new TransformingIterator<>(
                                        dataProvider.iterator(),
                                        row -> project(row, requiredColumns)
                                )
                        ),
                        rowFactory::create
                );
            }

            @Override
            public <RowT> Publisher<RowT> indexLookup(ExecutionContext<RowT> ctx, PartitionWithConsistencyToken partWithConsistencyToken,
                    RowFactory<RowT> rowFactory, int indexId, List<String> columns, RowT key, @Nullable BitSet requiredColumns) {
                throw new UnsupportedOperationException();
            }

            @Override
            public <RowT> CompletableFuture<@Nullable RowT> primaryKeyLookup(ExecutionContext<RowT> ctx, InternalTransaction tx,
                    RowFactory<RowT> rowFactory, RowT key, @Nullable BitSet requiredColumns) {
                throw new UnsupportedOperationException();
            }
        };
    }

    /**
     * Factory method to create {@link ScannableTable table} instance from given data provider with
     * only implemented {@link ScannableTable#indexLookup index lookup}.
     */
    public static ScannableTable indexLookup(DataProvider<Object[]> dataProvider) {
        return new ScannableTable() {
            @Override
            public <RowT> Publisher<RowT> scan(
                    ExecutionContext<RowT> ctx,
                    PartitionWithConsistencyToken partWithConsistencyToken,
                    RowFactory<RowT> rowFactory,
                    @Nullable BitSet requiredColumns
            ) {
                throw new UnsupportedOperationException();
            }

            @Override
            public <RowT> Publisher<RowT> indexRangeScan(ExecutionContext<RowT> ctx, PartitionWithConsistencyToken partWithConsistencyToken,
                    RowFactory<RowT> rowFactory, int indexId, List<String> columns, @Nullable RangeCondition<RowT> cond,
                    @Nullable BitSet requiredColumns) {
                throw new UnsupportedOperationException();
            }

            @Override
            public <RowT> Publisher<RowT> indexLookup(ExecutionContext<RowT> ctx, PartitionWithConsistencyToken partWithConsistencyToken,
                    RowFactory<RowT> rowFactory, int indexId, List<String> columns, RowT key, @Nullable BitSet requiredColumns) {
                return new TransformingPublisher<>(
                        SubscriptionUtils.fromIterable(
                                () -> new TransformingIterator<>(
                                        dataProvider.iterator(),
                                        row -> project(row, requiredColumns)
                                )
                        ),
                        rowFactory::create
                );
            }

            @Override
            public <RowT> CompletableFuture<@Nullable RowT> primaryKeyLookup(ExecutionContext<RowT> ctx, InternalTransaction tx,
                    RowFactory<RowT> rowFactory, RowT key, @Nullable BitSet requiredColumns) {
                throw new UnsupportedOperationException();
            }
        };
    }

    /**
     * A builder to create a test cluster object.
     *
     * @see TestCluster
     */
    public interface ClusterBuilder {
        /**
         * Sets desired names for the cluster nodes.
         *
         * @param firstNodeName A name of the first node. There is no difference in what node should be first. This parameter was
         *         introduced to force user to provide at least one node name.
         * @param otherNodeNames An array of rest of the names to create cluster from.
         * @return {@code this} for chaining.
         */
        ClusterBuilder nodes(String firstNodeName, String... otherNodeNames);

        /**
         * Sets desired names for the cluster nodes.
         *
         * @param firstNodeName A name of the first node. There is no difference in what node should be first. This parameter was
         *         introduced to force user to provide at least one node name.
         * @param useTablePartitions If {@code true} map table partitions to whole defined nodes.
         * @param otherNodeNames An array of rest of the names to create cluster from.
         * @return {@code this} for chaining.
         */
        public ClusterBuilder nodes(String firstNodeName, boolean useTablePartitions, String... otherNodeNames);

        /**
         * Creates a table builder to add to the cluster.
         *
         * @return An instance of table builder.
         */
        ClusterTableBuilder addTable();

        /**
         * Adds the given system view to the cluster.
         *
         * @param systemView System view.
         * @return {@code this} for chaining.
         * @param <T> System view data type.
         */
        <T> ClusterBuilder addSystemView(SystemView<T> systemView);

        /**
         * Builds the cluster object.
         *
         * @return Created cluster object.
         */
        TestCluster build();

        /**
         * Provides implementation of table with given name local per given node.
         *
         * @param nodeName Name of the node given instance of table will be assigned to.
         * @param tableName Name of the table given instance represents.
         * @param table Actual table that will be used for read operations during execution.
         * @return {@code this} for chaining.
         */
        ClusterBuilder dataProvider(String nodeName, String tableName, ScannableTable table);

        /**
         * Registers a previously added system view (see {@link #addSystemView(SystemView)}) on the specified node.
         *
         * @param nodeName Name of the node the view is going to be registered at.
         * @param systemViewName Name of previously registered system.
         * @return {@code this} for chaining.
         */
        ClusterBuilder registerSystemView(String nodeName, String systemViewName);
    }

    /**
     * A builder to create a test table object.
     */
    public interface TableBuilder extends TableBuilderBase<TableBuilder> {
        /** Returns a builder of the test sorted-index object. */
        SortedIndexBuilder sortedIndex();

        /** Returns a builder of the test hash-index object. */
        HashIndexBuilder hashIndex();

        /** Sets the distribution of the table. */
        TableBuilder distribution(IgniteDistribution distribution);

        /** Sets the size of the table. */
        TableBuilder size(int size);

        /** Sets id for the table. The caller must guarantee that provided id is unique. */
        TableBuilder tableId(int id);

        /** Sets the number of partitions fot this table. Default value is equal to {@link CatalogUtils#DEFAULT_PARTITION_COUNT}. */
        TableBuilder partitions(int num);

        /**
         * Builds a table.
         *
         * @return Created table object.
         */
        IgniteTable build();
    }

    /**
     * A builder to create a test object that representing sorted index.
     *
     * @see TestIndex
     */
    public interface SortedIndexBuilder extends SortedIndexBuilderBase<SortedIndexBuilder>, NestedBuilder<TableBuilder> {
    }

    /**
     * A builder to create a test object that representing hash index.
     *
     * @see TestIndex
     */
    public interface HashIndexBuilder extends HashIndexBuilderBase<HashIndexBuilder>, NestedBuilder<TableBuilder> {
    }

    /**
     * A builder to create a test table as nested object of the cluster.
     *
     * @see TestCluster
     */
    public interface ClusterTableBuilder extends TableBuilderBase<ClusterTableBuilder>,
            NestedBuilder<ClusterBuilder> {

        /**
         * Creates a sorted-index builder to add to the cluster.
         *
         * @return An instance of sorted-index builder.
         */
        ClusterSortedIndexBuilder addSortedIndex();

        /**
         * Creates a hash-index builder to add to the cluster.
         *
         * @return An instance of hash builder.
         */
        ClusterHashIndexBuilder addHashIndex();
    }

    /**
     * A builder to create a test object, which represents sorted index, as nested object of the cluster.
     *
     * @see TestIndex
     * @see TestCluster
     */
    public interface ClusterSortedIndexBuilder extends SortedIndexBuilderBase<ClusterSortedIndexBuilder>,
            NestedBuilder<ClusterTableBuilder> {
    }

    /**
     * A builder to create a test object, which represents hash index, as nested object of the cluster.
     *
     * @see TestIndex
     * @see TestCluster
     */
    public interface ClusterHashIndexBuilder extends HashIndexBuilderBase<ClusterHashIndexBuilder>,
            NestedBuilder<ClusterTableBuilder> {
    }

    /**
     * A builder to create an execution context.
     *
     * @see ExecutionContext
     */
    public interface ExecutionContextBuilder {
        /** Sets the identifier of the query. */
        ExecutionContextBuilder queryId(UUID queryId);

        /** Sets the description of the fragment this context will be created for. */
        ExecutionContextBuilder fragment(FragmentDescription fragmentDescription);

        /** Sets the query task executor. */
        ExecutionContextBuilder executor(QueryTaskExecutor executor);

        /** Sets the node this fragment will be executed on. */
        ExecutionContextBuilder localNode(ClusterNode node);

        /** Sets the dynamic parameters this fragment will be executed with. */
        ExecutionContextBuilder dynamicParameters(Object... params);

        /**
         * Builds the context object.
         *
         * @return Created context object.
         */
        ExecutionContext<Object[]> build();
    }

    private static class ExecutionContextBuilderImpl implements ExecutionContextBuilder {
        private FragmentDescription description = new FragmentDescription(0, true, null, null, null, null);

        private UUID queryId = null;
        private QueryTaskExecutor executor = null;
        private ClusterNode node = null;
        private Object[] dynamicParams = ArrayUtils.OBJECT_EMPTY_ARRAY;

        /** {@inheritDoc} */
        @Override
        public ExecutionContextBuilder queryId(UUID queryId) {
            this.queryId = Objects.requireNonNull(queryId, "queryId");

            return this;
        }

        /** {@inheritDoc} */
        @Override
        public ExecutionContextBuilder fragment(FragmentDescription fragmentDescription) {
            this.description = Objects.requireNonNull(fragmentDescription, "fragmentDescription");

            return this;
        }

        /** {@inheritDoc} */
        @Override
        public ExecutionContextBuilder executor(QueryTaskExecutor executor) {
            this.executor = Objects.requireNonNull(executor, "executor");

            return this;
        }

        /** {@inheritDoc} */
        @Override
        public ExecutionContextBuilder localNode(ClusterNode node) {
            this.node = Objects.requireNonNull(node, "node");

            return this;
        }

        @Override
        public ExecutionContextBuilder dynamicParameters(Object... params) {
            this.dynamicParams = params;
            return this;
        }

        /** {@inheritDoc} */
        @Override
        public ExecutionContext<Object[]> build() {
            return new ExecutionContext<>(
                    Objects.requireNonNull(executor, "executor"),
                    queryId,
                    Objects.requireNonNull(node, "node"),
                    node.name(),
                    description,
                    ArrayRowHandler.INSTANCE,
                    Commons.parametersMap(dynamicParams),
                    TxAttributes.fromTx(new NoOpTransaction(node.name())),
                    SqlQueryProcessor.DEFAULT_TIME_ZONE_ID
            );
        }
    }

    private static class ClusterBuilderImpl implements ClusterBuilder {
        private final List<ClusterTableBuilderImpl> tableBuilders = new ArrayList<>();
        private List<String> nodeNames;
        private boolean useTablePartitions;
        private final Map<String, Map<String, ScannableTable>> nodeName2tableName2table = new HashMap<>();
        private final List<SystemView<?>> systemViews = new ArrayList<>();
        private final Map<String, Set<String>> nodeName2SystemView = new HashMap<>();

        /** {@inheritDoc} */
        @Override
        public ClusterBuilder nodes(String firstNodeName, String... otherNodeNames) {
            this.nodeNames = new ArrayList<>();

            nodeNames.add(firstNodeName);
            nodeNames.addAll(Arrays.asList(otherNodeNames));

            return this;
        }

        /** {@inheritDoc} */
        @Override
        public ClusterBuilder nodes(String firstNodeName, boolean useTablePartitions, String... otherNodeNames) {
            this.nodeNames = new ArrayList<>();
            this.useTablePartitions = useTablePartitions;

            nodeNames.add(firstNodeName);
            nodeNames.addAll(Arrays.asList(otherNodeNames));

            return this;
        }

        /** {@inheritDoc} */
        @Override
        public ClusterTableBuilder addTable() {
            return new ClusterTableBuilderImpl(this);
        }

        @Override
        public <T> ClusterBuilder addSystemView(SystemView<T> systemView) {
            systemViews.add(systemView);
            return this;
        }

        @Override
        public ClusterBuilder dataProvider(String nodeName, String tableName, ScannableTable table) {
            nodeName2tableName2table.computeIfAbsent(nodeName, key -> new HashMap<>()).put(tableName, table);

            return this;
        }

        @Override
        public ClusterBuilder registerSystemView(String nodeName, String systemViewName) {
            nodeName2SystemView.computeIfAbsent(nodeName, key -> new HashSet<>()).add(systemViewName);

            return this;
        }

        /** {@inheritDoc} */
        @Override
        public TestCluster build() {
            validateConfiguredDataProviders();

            var clusterService = new ClusterServiceFactory(nodeNames);

            var clusterName = "test_cluster";

            HybridClock clock = new HybridClockImpl();
            CatalogManager catalogManager = CatalogTestUtils.createCatalogManagerWithTestUpdateLog(clusterName, clock);

            var parserService = new ParserServiceImpl();
            var prepareService = new PrepareServiceImpl(clusterName, 0, CaffeineCacheFactory.INSTANCE,
                    new DdlSqlToCommandConverter(), PLANNING_TIMEOUT, PLANNING_THREAD_COUNT,
                    mock(MetricManagerImpl.class));

            Map<String, List<String>> owningNodesByTableName = new HashMap<>();
            for (Entry<String, Map<String, ScannableTable>> entry : nodeName2tableName2table.entrySet()) {
                for (String tableName : entry.getValue().keySet()) {
                    owningNodesByTableName.computeIfAbsent(tableName, key -> new ArrayList<>()).add(entry.getKey());
                }
            }

            Map<String, List<String>> systemViewsByNode = new HashMap<>();

            for (Entry<String, Set<String>> entry : nodeName2SystemView.entrySet()) {
                String nodeName = entry.getKey();
                for (String systemViewName : entry.getValue()) {
                    systemViewsByNode.computeIfAbsent(nodeName, (k) -> new ArrayList<>()).add(systemViewName);
                }
            }

            ClockWaiter clockWaiter = new ClockWaiter("test", clock);
            var ddlHandler = new DdlCommandHandler(catalogManager, new TestClockService(clock, clockWaiter), () -> 100);
            var schemaManager = new SqlSchemaManagerImpl(catalogManager, CaffeineCacheFactory.INSTANCE, 0);

            Runnable initClosure = () -> {
                assertThat(clockWaiter.startAsync(), willCompleteSuccessfully());

                initAction(catalogManager);
            };

            List<LogicalNode> logicalNodes = nodeNames.stream()
                    .map(name -> {
                        List<String> systemViewForNode = systemViewsByNode.getOrDefault(name, List.of());
                        NetworkAddress addr = NetworkAddress.from("127.0.0.1:10000");
                        LogicalNode logicalNode = new LogicalNode(name, name, addr);

                        if (systemViewForNode.isEmpty()) {
                            return logicalNode;
                        } else {
                            String attrName = SystemViewManagerImpl.NODE_ATTRIBUTES_KEY;
                            String nodeNameSep = SystemViewManagerImpl.NODE_ATTRIBUTES_LIST_SEPARATOR;
                            String nodeNamesString = String.join(nodeNameSep, systemViewForNode);

                            return new LogicalNode(logicalNode, Map.of(), Map.of(attrName, nodeNamesString), List.of());
                        }
                    })
                    .collect(Collectors.toList());

            Map<String, TestNode> nodes = nodeNames.stream()
                    .map(name -> {
                        var systemViewManager = new SystemViewManagerImpl(name, catalogManager);
                        var targetProvider = new TestNodeExecutionTargetProvider(
                                systemViewManager::owningNodes,
                                owningNodesByTableName,
                                useTablePartitions
                        );
                        var partitionPruner = new PartitionPrunerImpl();
                        var mappingService = new MappingServiceImpl(
                                name,
                                targetProvider,
                                EmptyCacheFactory.INSTANCE,
                                0,
                                partitionPruner,
                                Runnable::run
                        );

                        systemViewManager.register(() -> systemViews);

                        LogicalTopologySnapshot newTopology = new LogicalTopologySnapshot(1L, logicalNodes);
                        mappingService.onTopologyLeap(newTopology);
                        systemViewManager.onTopologyLeap(newTopology);

                        return new TestNode(
                                name,
                                clusterService.forNode(name),
                                parserService,
                                prepareService,
                                schemaManager,
                                mappingService,
                                new TestExecutableTableRegistry(nodeName2tableName2table.get(name), schemaManager),
                                ddlHandler,
                                systemViewManager
                        );
                    })
                    .collect(Collectors.toMap(TestNode::name, Function.identity()));

            return new TestCluster(
                    nodes,
                    catalogManager,
                    prepareService,
                    clockWaiter,
                    initClosure
            );
        }

        private void validateConfiguredDataProviders() {
            Set<String> dataProvidersOwners = new HashSet<>(nodeName2tableName2table.keySet());

            dataProvidersOwners.removeAll(Set.copyOf(nodeNames));

            if (!dataProvidersOwners.isEmpty()) {
                Map<String, List<String>> problematicTables = new HashMap<>();

                for (String outsiderNode : dataProvidersOwners) {
                    for (String problematicTable : nodeName2tableName2table.get(outsiderNode).keySet()) {
                        problematicTables.computeIfAbsent(problematicTable, k -> new ArrayList<>()).add(outsiderNode);
                    }
                }

                String problematicTablesString = problematicTables.entrySet().stream()
                        .map(e -> e.getKey() + ": " + e.getValue())
                        .collect(Collectors.joining(", "));

                throw new AssertionError(format("The table has a dataProvider that is outside the cluster "
                        + "[{}]", problematicTablesString));
            }
        }

        private void initAction(CatalogManager catalogManager) {
            List<CatalogCommand> initialSchema = tableBuilders.stream()
                    .flatMap(builder -> builder.build().stream())
                    .collect(Collectors.toList());

            CompletableFuture<Boolean> indicesReadyFut = new CompletableFuture<>();
            CopyOnWriteArraySet<Integer> initialIndices = new CopyOnWriteArraySet<>();

            // Make indices registered via builder API available on startup.
            if (!tableBuilders.isEmpty()) {
                Consumer<MakeIndexAvailableEventParameters> indexAvailableHandler = params -> {
                    initialIndices.remove(params.indexId());

                    if (initialIndices.isEmpty()) {
                        indicesReadyFut.complete(true);
                    }
                };

                EventListener<MakeIndexAvailableEventParameters> listener = EventListener.fromConsumer(indexAvailableHandler);
                catalogManager.listen(CatalogEvent.INDEX_AVAILABLE, listener);

                // Remove listener, when all indices become available.
                indicesReadyFut.whenComplete((r, t) -> {
                    catalogManager.removeListener(CatalogEvent.INDEX_AVAILABLE, listener);
                });
            } else {
                indicesReadyFut.complete(true);
            }

            // Every time an index is created add `start building `and `make available` commands
            // to make that index accessible to the SQL engine.
            Consumer<CreateIndexEventParameters> createIndexHandler = (params) -> {
                CatalogIndexDescriptor index = params.indexDescriptor();
                int indexId = index.id();

                CatalogCommand startBuildIndexCommand = StartBuildingIndexCommand.builder().indexId(indexId).build();
                CatalogCommand makeIndexAvailableCommand = MakeIndexAvailableCommand.builder().indexId(indexId).build();

                // Collect initial indexes only if catalog init future has not completed.
                if (!indicesReadyFut.isDone()) {
                    initialIndices.add(indexId);
                }

                LOG.info("Index has been created. Sending commands to make index available. id: {}, name: {}, status: {}",
                        indexId, index.name(), index.status());

                catalogManager.execute(List.of(startBuildIndexCommand, makeIndexAvailableCommand));
            };
            catalogManager.listen(CatalogEvent.INDEX_CREATE, EventListener.fromConsumer(createIndexHandler));

            // Init schema.
            await(catalogManager.execute(initialSchema));

            // Wait until all indices become available.
            await(indicesReadyFut);
        }
    }

    private static class TableBuilderImpl implements TableBuilder {
        private final List<AbstractTableIndexBuilderImpl<?>> indexBuilders = new ArrayList<>();
        private final List<ColumnDescriptor> columns = new ArrayList<>();

        private String name;
        private IgniteDistribution distribution;
        private int size = 100_000;
        private Integer tableId;
        private int partitions = CatalogUtils.DEFAULT_PARTITION_COUNT;

        /** {@inheritDoc} */
        @Override
        public TableBuilder name(String name) {
            this.name = name;

            return this;
        }

        /** {@inheritDoc} */
        @Override
        public SortedIndexBuilder sortedIndex() {
            return new SortedIndexBuilderImpl(this);
        }

        /** {@inheritDoc} */
        @Override
        public HashIndexBuilder hashIndex() {
            return new HashIndexBuilderImpl(this);
        }

        /** {@inheritDoc} */
        @Override
        public TableBuilder distribution(IgniteDistribution distribution) {
            this.distribution = distribution;

            return this;
        }

        /** {@inheritDoc} */
        @Override
        public TableBuilder addColumn(String name, NativeType type, boolean nullable) {
            columns.add(new ColumnDescriptorImpl(
                    name, false, false, nullable, columns.size(), type, DefaultValueStrategy.DEFAULT_NULL, null
            ));

            return this;
        }

        /** {@inheritDoc} */
        @Override
        public TableBuilder addColumn(String name, NativeType type) {
            return addColumn(name, type, true);
        }

        /** {@inheritDoc} */
        @Override
        public TableBuilder addColumn(String name, NativeType type, @Nullable Object defaultValue) {
            if (defaultValue == null) {
                return addColumn(name, type);
            } else {
                ColumnDescriptorImpl desc = new ColumnDescriptorImpl(
                        name, false, false, true, columns.size(), type, DefaultValueStrategy.DEFAULT_CONSTANT, () -> defaultValue
                );
                columns.add(desc);
            }

            return this;
        }

        /** {@inheritDoc} */
        @Override
        public TableBuilder addKeyColumn(String name, NativeType type) {
            columns.add(new ColumnDescriptorImpl(
                    name, true, false, false, columns.size(), type, DefaultValueStrategy.DEFAULT_NULL, null
            ));

            return this;
        }

        /** {@inheritDoc} */
        @Override
        public TableBuilder size(int size) {
            this.size = size;

            return this;
        }

        /** {@inheritDoc} */
        @Override
        public TableBuilder tableId(int id) {
            this.tableId = id;

            return this;
        }

        /** {@inheritDoc} */
        @Override
        public TableBuilder partitions(int num) {
            this.partitions = num;
            return this;
        }

        /** {@inheritDoc} */
        @Override
        public IgniteTable build() {
            if (distribution == null) {
                throw new IllegalArgumentException("Distribution is not specified");
            }

            if (name == null) {
                throw new IllegalArgumentException("Name is not specified");
            }

            if (columns.isEmpty()) {
                throw new IllegalArgumentException("Table must contain at least one column");
            }

            TableDescriptorImpl tableDescriptor = new TableDescriptorImpl(columns, distribution);

            Map<String, IgniteIndex> indexes = indexBuilders.stream()
                    .map(idx -> idx.build(tableDescriptor))
                    .collect(Collectors.toUnmodifiableMap(IgniteIndex::name, Function.identity()));

            return new IgniteTableImpl(
                    Objects.requireNonNull(name),
                    tableId != null ? tableId : TABLE_ID_GEN.incrementAndGet(),
                    1,
                    tableDescriptor,
                    findPrimaryKey(tableDescriptor, indexes.values()),
                    new TestStatistic(size),
                    indexes,
                    partitions
            );
        }
    }

    private static ImmutableIntList findPrimaryKey(TableDescriptor descriptor, Collection<IgniteIndex> indexList) {
        IgniteIndex primaryKey = indexList.stream()
                .filter(IgniteIndex::primaryKey)
                .findFirst()
                .orElse(null);

        if (primaryKey != null) {
            return primaryKey.collation().getKeys();
        }

        List<Integer> list = new ArrayList<>();
        for (ColumnDescriptor column : descriptor) {
            if (column.key()) {
                list.add(column.logicalIndex());
            }
        }

        return ImmutableIntList.copyOf(list);
    }

    private static class ClusterTableBuilderImpl implements ClusterTableBuilder {
        private final List<AbstractClusterTableIndexBuilderImpl<?>> indexBuilders = new ArrayList<>();

        private final List<ColumnParams> columns = new ArrayList<>();
        private final List<String> keyColumns = new ArrayList<>();

        private final ClusterBuilderImpl parent;

        private final String schemaName = CatalogManager.DEFAULT_SCHEMA_NAME;

        private String name;

        private ClusterTableBuilderImpl(ClusterBuilderImpl parent) {
            this.parent = parent;
        }

        /** {@inheritDoc} */
        @Override
        public ClusterTableBuilder name(String name) {
            this.name = name;

            return this;
        }

        /** {@inheritDoc} */
        @Override
        public ClusterSortedIndexBuilder addSortedIndex() {
            return new ClusterSortedIndexBuilderImpl(this);
        }

        /** {@inheritDoc} */
        @Override
        public ClusterHashIndexBuilder addHashIndex() {
            return new ClusterHashIndexBuilderImpl(this);
        }

        @Override
        public ClusterTableBuilder addColumn(String name, NativeType type, boolean nullable) {
            columns.add(columnParams(name, type, nullable, null));

            return this;
        }

        /** {@inheritDoc} */
        @Override
        public ClusterTableBuilder addColumn(String name, NativeType type) {
            return addColumn(name, type, true);
        }

        @Override
        public ClusterTableBuilder addColumn(String name, NativeType type, @Nullable Object defaultValue) {
            columns.add(columnParams(name, type, true, defaultValue));

            return this;
        }

        @Override
        public ClusterTableBuilder addKeyColumn(String name, NativeType type) {
            keyColumns.add(name);

            return addColumn(name, type, false);
        }

        /** {@inheritDoc} */
        @Override
        public ClusterBuilder end() {
            parent.tableBuilders.add(this);

            return parent;
        }

        private List<CatalogCommand> build() {
            List<CatalogCommand> commands = new ArrayList<>(1 + indexBuilders.size());

            // TODO https://issues.apache.org/jira/browse/IGNITE-21715 Update after TestFramework provides API
            //  to specify type of a primary key index.
            // Use sorted index by default.
            TablePrimaryKey primaryKey = TableHashPrimaryKey.builder()
                    .columns(keyColumns)
                    .build();

            commands.add(
                    CreateTableCommand.builder()
                            .schemaName(schemaName)
                            .tableName(name)
                            .columns(columns)
                            .primaryKey(primaryKey)
                            .build()
            );

            for (AbstractClusterTableIndexBuilderImpl<?> builder : indexBuilders) {
                commands.add(builder.build(schemaName, name));
            }

            return commands;
        }
    }

    private static class SortedIndexBuilderImpl extends AbstractTableIndexBuilderImpl<SortedIndexBuilder>
            implements SortedIndexBuilder {
        private final TableBuilderImpl parent;

        private SortedIndexBuilderImpl(TableBuilderImpl parent) {
            this.parent = parent;
        }

        /** {@inheritDoc} */
        @Override
        SortedIndexBuilder self() {
            return this;
        }

        /** {@inheritDoc} */
        @Override
        public TableBuilder end() {
            parent.indexBuilders.add(this);

            return parent;
        }

        /** {@inheritDoc} */
        @Override
        public TestIndex build(TableDescriptor desc) {
            if (name == null) {
                throw new IllegalArgumentException("Name is not specified");
            }

            if (columns.isEmpty()) {
                throw new IllegalArgumentException("Index must contain at least one column");
            }

            if (collations.size() != columns.size()) {
                throw new IllegalArgumentException("Collation must be specified for each of columns.");
            }

            return TestIndex.createSorted(name, columns, collations, desc);
        }
    }

    private static class HashIndexBuilderImpl extends AbstractTableIndexBuilderImpl<HashIndexBuilder> implements HashIndexBuilder {
        private final TableBuilderImpl parent;

        private HashIndexBuilderImpl(TableBuilderImpl parent) {
            this.parent = parent;
        }

        /** {@inheritDoc} */
        @Override
        HashIndexBuilder self() {
            return this;
        }

        /** {@inheritDoc} */
        @Override
        public TableBuilder end() {
            parent.indexBuilders.add(this);

            return parent;
        }

        /** {@inheritDoc} */
        @Override
        public TestIndex build(TableDescriptor desc) {
            if (name == null) {
                throw new IllegalArgumentException("Name is not specified");
            }

            if (columns.isEmpty()) {
                throw new IllegalArgumentException("Index must contain at least one column");
            }

            assert collations == null : "Collation is not supported.";

            return TestIndex.createHash(name, columns, desc);
        }
    }

    private static class ClusterSortedIndexBuilderImpl extends AbstractClusterTableIndexBuilderImpl<ClusterSortedIndexBuilder>
            implements ClusterSortedIndexBuilder {
        private final ClusterTableBuilderImpl parent;

        ClusterSortedIndexBuilderImpl(ClusterTableBuilderImpl parent) {
            this.parent = parent;
        }

        /** {@inheritDoc} */
        @Override
        ClusterSortedIndexBuilder self() {
            return this;
        }

        /** {@inheritDoc} */
        @Override
        public ClusterTableBuilder end() {
            parent.indexBuilders.add(this);

            return parent;
        }

        @Override
        CatalogCommand build(String schemaName, String tableName) {
            assert collations.size() == columns.size();

            List<CatalogColumnCollation> catalogCollations = collations.stream()
                    .map(c -> CatalogColumnCollation.get(c.asc, c.nullsFirst))
                    .collect(Collectors.toList());

            return CreateSortedIndexCommand.builder()
                    .schemaName(schemaName)
                    .tableName(tableName)
                    .indexName(name)
                    .columns(columns)
                    .collations(catalogCollations)
                    .build();
        }
    }

    private static class ClusterHashIndexBuilderImpl extends AbstractClusterTableIndexBuilderImpl<ClusterHashIndexBuilder>
            implements ClusterHashIndexBuilder {
        private final ClusterTableBuilderImpl parent;

        ClusterHashIndexBuilderImpl(ClusterTableBuilderImpl parent) {
            this.parent = parent;
        }

        /** {@inheritDoc} */
        @Override
        ClusterHashIndexBuilder self() {
            return this;
        }

        /** {@inheritDoc} */
        @Override
        public ClusterTableBuilder end() {
            parent.indexBuilders.add(this);

            return parent;
        }

        @Override
        CatalogCommand build(String schemaName, String tableName) {
            return CreateHashIndexCommand.builder()
                    .schemaName(schemaName)
                    .tableName(tableName)
                    .indexName(name)
                    .columns(columns)
                    .build();
        }
    }

    private abstract static class AbstractIndexBuilderImpl<ChildT> implements SortedIndexBuilderBase<ChildT>, HashIndexBuilderBase<ChildT> {
        String name;
        final List<String> columns = new ArrayList<>();
        List<Collation> collations;

        /** {@inheritDoc} */
        @Override
        public ChildT name(String name) {
            this.name = name;

            return self();
        }

        /** {@inheritDoc} */
        @Override
        public ChildT addColumn(String columnName) {
            columns.add(columnName);

            return self();
        }

        /** {@inheritDoc} */
        @Override
        public ChildT addColumn(String columnName, Collation collation) {
            if (collations == null) {
                collations = new ArrayList<>();
            }

            columns.add(columnName);
            collations.add(collation);

            return self();
        }

        abstract ChildT self();
    }

    private abstract static class AbstractTableIndexBuilderImpl<ChildT> extends AbstractIndexBuilderImpl<ChildT> {
        abstract TestIndex build(TableDescriptor desc);
    }

    private abstract static class AbstractClusterTableIndexBuilderImpl<ChildT> extends AbstractIndexBuilderImpl<ChildT> {
        abstract CatalogCommand build(String schemaName, String tableName);
    }

    /**
     * Base interface describing the complete set of table-related fields.
     *
     * <p>The sole purpose of this interface is to keep in sync both variants of table's builders.
     *
     * @param <ChildT> An actual type of builder that should be exposed to the user.
     * @see ClusterTableBuilder
     * @see TableBuilder
     */
    private interface TableBuilderBase<ChildT> {
        /** Sets the name of the table. */
        ChildT name(String name);

        /** Adds a key column to the table. */
        ChildT addKeyColumn(String name, NativeType type);

        /** Adds a column to the table. */
        ChildT addColumn(String name, NativeType type);

        /** Adds a column with given nullability to the table. */
        ChildT addColumn(String name, NativeType type, boolean nullable);

        /** Adds a column with the given default value to the table. */
        ChildT addColumn(String name, NativeType type, @Nullable Object defaultValue);
    }

    /**
     * Base interface describing the common set of index-related fields.
     *
     * <p>The sole purpose of this interface is to keep in sync both variants of index's builders.
     *
     * @param <ChildT> An actual type of builder that should be exposed to the user.
     * @see ClusterHashIndexBuilder
     * @see ClusterSortedIndexBuilder
     * @see HashIndexBuilder
     * @see SortedIndexBuilder
     */
    private interface IndexBuilderBase<ChildT> {
        /** Sets the name of the index. */
        ChildT name(String name);
    }

    /**
     * Base interface describing the set of sorted-index related fields.
     *
     * @param <ChildT> An actual type of builder that should be exposed to the user.
     */
    private interface SortedIndexBuilderBase<ChildT> extends IndexBuilderBase<ChildT> {
        /** Adds a column with specified collation to the index. */
        ChildT addColumn(String columnName, Collation collation);
    }

    /**
     * Base interface describing the set of hash-index related fields.
     *
     * @param <ChildT> An actual type of builder that should be exposed to the user.
     */
    private interface HashIndexBuilderBase<ChildT> extends IndexBuilderBase<ChildT> {
        /** Adds a column to the index. */
        ChildT addColumn(String columnName);
    }

    /**
     * This interfaces provides a nested builder with ability to return on the previous layer.
     *
     * <p>For example:</p>
     * <pre>
     *     interface ChildBuilder implements NestedBuilder&lt;ParentBuilder&gt; {
     *         ChildBuilder nestedFoo();
     *     }
     *
     *     interface ParentBuilder {
     *         ParentBuilder foo();
     *         ParentBuilder bar();
     *         ChildBuilder child();
     *     }
     *
     *     Builders.parent()
     *         .foo()
     *         .child() // now we are dealing with the ChildBuilder
     *             .nestedFoo()
     *             .end() // and here we are returning back to the ParentBuilder
     *         .bar()
     *         .build()
     * </pre>
     */
    @FunctionalInterface
    private interface NestedBuilder<ParentT> {
        /**
         * Notifies the builder's chain of the nested builder that we need to return back to the previous layer.
         *
         * @return An instance of the parent builder.
         */
        ParentT end();
    }

    private static class TestExecutableTableRegistry implements ExecutableTableRegistry {
        private final Map<String, ScannableTable> tablesByName;
        private final SqlSchemaManager schemaManager;

        TestExecutableTableRegistry(Map<String, ScannableTable> tablesByName, SqlSchemaManager schemaManager) {
            this.tablesByName = tablesByName;
            this.schemaManager = schemaManager;
        }

        @Override
        public CompletableFuture<ExecutableTable> getTable(int schemaVersion, int tableId) {
            IgniteTable table = schemaManager.table(schemaVersion, tableId);

            assert table != null;

            return CompletableFuture.completedFuture(new ExecutableTable() {
                @Override
                public ScannableTable scannableTable() {
                    ScannableTable scannableTable = tablesByName.get(table.name());

                    assert scannableTable != null;

                    return scannableTable;
                }

                @Override
                public UpdatableTable updatableTable() {
                    throw new UnsupportedOperationException();
                }

                @Override
                public TableDescriptor tableDescriptor() {
                    return table.descriptor();
                }

                @Override
                public Supplier<PartitionCalculator> partitionCalculator() {
                    return table.partitionCalculator();
                }
            });
        }
    }

    private static ColumnParams columnParams(String name, NativeType type, boolean nullable, @Nullable Object defaultValue) {
        NativeTypeSpec typeSpec = type.spec();

        Builder builder = ColumnParams.builder()
                .name(name)
                .type(typeSpec.asColumnType())
                .nullable(nullable)
                .defaultValue(DefaultValue.constant(defaultValue));

        switch (typeSpec) {
            case INT8:
            case INT16:
            case INT32:
            case INT64:
            case FLOAT:
            case DOUBLE:
            case DATE:
            case UUID:
            case BOOLEAN:
                break;
            case NUMBER:
                assert type instanceof NumberNativeType : type.getClass().getCanonicalName();

                builder.precision(((NumberNativeType) type).precision());
                break;
            case DECIMAL:
                assert type instanceof DecimalNativeType : type.getClass().getCanonicalName();

                builder.precision(((DecimalNativeType) type).precision());
                builder.scale(((DecimalNativeType) type).scale());
                break;
            case STRING:
            case BYTES:
                assert type instanceof VarlenNativeType : type.getClass().getCanonicalName();

                builder.length(((VarlenNativeType) type).length());
                break;
            case BITMASK:
                assert type instanceof BitmaskNativeType : type.getClass().getCanonicalName();

                builder.length(((BitmaskNativeType) type).bits());
                break;
            case TIME:
            case DATETIME:
            case TIMESTAMP:
                assert type instanceof TemporalNativeType : type.getClass().getCanonicalName();

                builder.precision(((TemporalNativeType) type).precision());
                break;
            default:
                throw new IllegalArgumentException("Unsupported native type: " + typeSpec);
        }

        return builder.build();
    }

    private static Object[] project(Object[] row, @Nullable BitSet requiredElements) {
        if (requiredElements == null) {
            return row;
        }

        Object[] newRow = new Object[requiredElements.cardinality()];

        int idx = 0;
        for (int i = requiredElements.nextSetBit(0); i != -1; i = requiredElements.nextSetBit(i + 1)) {
            newRow[idx++] = row[i];
        }

        return newRow;
    }

    /** Returns a builder for {@link ExecutionTargetProvider}. */
    public static ExecutionTargetProviderBuilder executionTargetProviderBuilder() {
        return new ExecutionTargetProviderBuilder();
    }

    /** A builder to create instances of {@link ExecutionTargetProvider}. */
    public static final class ExecutionTargetProviderBuilder {

        private final Map<String, List<String>> owningNodesByTableName = new HashMap<>();

        private Function<String, List<String>> owningNodesBySystemViewName = (n) -> null;

        private boolean useTablePartitions;

        private ExecutionTargetProviderBuilder() {

        }

        /** Adds tables to list of nodes mapping. */
        public ExecutionTargetProviderBuilder addTables(Map<String, List<String>> tables) {
            this.owningNodesByTableName.putAll(tables);
            return this;
        }

        /**
         * Sets a function that returns system views. Function accepts a view name and returns a list of nodes
         * a system view is available at.
         */
        public ExecutionTargetProviderBuilder setSystemViews(Function<String, List<String>> systemViews) {
            this.owningNodesBySystemViewName = systemViews;
            return this;
        }

        /** Use table partitions to build mapping targets. Default is {@code false}. */
        public ExecutionTargetProviderBuilder useTablePartitions(boolean value) {
            useTablePartitions = value;
            return this;
        }

        /** Creates an instance of {@link ExecutionTargetProvider}. */
        public ExecutionTargetProvider build() {
            return new TestNodeExecutionTargetProvider(
                    owningNodesBySystemViewName,
                    Map.copyOf(owningNodesByTableName),
                    useTablePartitions
            );
        }
    }

    private static class TestNodeExecutionTargetProvider implements ExecutionTargetProvider {

        final Function<String, List<String>> owningNodesBySystemViewName;

        final Map<String, List<String>> owningNodesByTableName;

        final boolean useTablePartitions;

        private TestNodeExecutionTargetProvider(
                Function<String, List<String>> owningNodesBySystemViewName,
                Map<String, List<String>> owningNodesByTableName,
                boolean useTablePartitions
        ) {
            this.owningNodesBySystemViewName = owningNodesBySystemViewName;
            this.owningNodesByTableName = Map.copyOf(owningNodesByTableName);
            this.useTablePartitions = useTablePartitions;
        }

        @Override
        public CompletableFuture<ExecutionTarget> forTable(ExecutionTargetFactory factory, IgniteTable table) {
            List<String> owningNodes = owningNodesByTableName.get(table.name());

            if (nullOrEmpty(owningNodes)) {
                throw new AssertionError("DataProvider is not configured for table " + table.name());
            }

            List<NodeWithConsistencyToken> nodes;

            if (useTablePartitions) {
                int p = table.partitions();

                nodes = IntStream.range(0, p).mapToObj(n -> {
                    String nodeName = owningNodes.get(n % owningNodes.size());
                    return new NodeWithConsistencyToken(nodeName, p);
                }).collect(Collectors.toList());
            } else {
                nodes = owningNodes.stream()
                        .map(name -> new NodeWithConsistencyToken(name, 1))
                        .collect(Collectors.toList());
            }

            ExecutionTarget target = factory.partitioned(nodes);

            return CompletableFuture.completedFuture(target);
        }

        @Override
        public CompletableFuture<ExecutionTarget> forSystemView(ExecutionTargetFactory factory, IgniteSystemView view) {
            List<String> nodes = owningNodesBySystemViewName.apply(view.name());

            if (nullOrEmpty(nodes)) {
                return CompletableFuture.failedFuture(
                        new SqlException(Sql.MAPPING_ERR, format("The view with name '{}' could not be found on"
                                + " any active nodes in the cluster", view.name()))
                );
            }

            return CompletableFuture.completedFuture(
                    view.distribution() == IgniteDistributions.single()
                            ? factory.oneOf(nodes)
                            : factory.allOf(nodes)
            );
        }
    }
}