[core] Support chain tbl on batch mode

Co-authored-by: zhoufa <[email protected]>

Author: yangmu.0722

docs/layouts/shortcodes/generated/core_configuration.html

@@ -98,6 +98,12 @@
             <td>MemorySize</td>
             <td>Memory page size for caching.</td>
         </tr>
+        <tr>
+            <td><h5>chain-table.enabled</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Specify chain table enable.</td>
+        </tr>
         <tr>
             <td><h5>changelog-file.compression</h5></td>
             <td style="word-wrap: break-word;">(none)</td>
@@ -1001,6 +1007,18 @@
             <td>String</td>
             <td>When a batch job queries from a table, if a partition does not exist in the current branch, the reader will try to get this partition from this fallback branch.</td>
         </tr>
+        <tr>
+            <td><h5>scan.fallback-delta-branch</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>String</td>
+            <td>When a batch job queries from a chain table, if a partition does not exist in the main and snapshot branch, the reader will try to get this partition from chain snapshot and delta branch together.</td>
+        </tr>
+        <tr>
+            <td><h5>scan.fallback-snapshot-branch</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>String</td>
+            <td>When a batch job queries from a chain table, if a partition does not exist in the main branch, the reader will try to get this partition from chain snapshot branch.</td>
+        </tr>
         <tr>
             <td><h5>scan.file-creation-time-millis</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -226,6 +226,35 @@ public InlineElement getDescription() {
     public static final ConfigOption<String> BRANCH =
             key("branch").stringType().defaultValue("main").withDescription("Specify branch name.");
 
+    public static final ConfigOption<Boolean> CHAIN_TABLE_ENABLED =
+            key("chain-table.enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription("Specify chain table enable.");
+
+    public static final ConfigOption<String> SCAN_FALLBACK_SNAPSHOT_BRANCH =
+            key("scan.fallback-snapshot-branch")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "When a batch job queries from a chain table, if a partition does not exist in the main branch, "
+                                    + "the reader will try to get this partition from chain snapshot branch.");
+
+    public static final ConfigOption<String> SCAN_FALLBACK_DELTA_BRANCH =
+            key("scan.fallback-delta-branch")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "When a batch job queries from a chain table, if a partition does not exist in the main and snapshot branch, "
+                                    + "the reader will try to get this partition from chain snapshot and delta branch together.");
+
+    @ExcludeFromDocumentation("Internal use only")
+    public static final ConfigOption<ChainBranchReadMode> CHAIN_TABLE_BRANCH_INTERNAL_READ_MODE =
+            key("chain-table.branch.internal.read.mode")
+                    .enumType(ChainBranchReadMode.class)
+                    .defaultValue(ChainBranchReadMode.DEFAULT)
+                    .withDescription("Chain query type.");
+
     public static final String FILE_FORMAT_ORC = "orc";
     public static final String FILE_FORMAT_AVRO = "avro";
     public static final String FILE_FORMAT_PARQUET = "parquet";
@@ -3131,6 +3160,22 @@ public int lookupMergeRecordsThreshold() {
         return options.get(LOOKUP_MERGE_RECORDS_THRESHOLD);
     }
 
+    public boolean isChainTable() {
+        return options.get(CHAIN_TABLE_ENABLED);
+    }
+
+    public String scanFallbackSnapshotBranch() {
+        return options.get(SCAN_FALLBACK_SNAPSHOT_BRANCH);
+    }
+
+    public String scanFallbackDeltaBranch() {
+        return options.get(SCAN_FALLBACK_DELTA_BRANCH);
+    }
+
+    public ChainBranchReadMode getChainBranchReadMode() {
+        return options.get(CHAIN_TABLE_BRANCH_INTERNAL_READ_MODE);
+    }
+
     public boolean formatTableImplementationIsPaimon() {
         return options.get(FORMAT_TABLE_IMPLEMENTATION) == FormatTableImplementation.PAIMON;
     }
@@ -3957,4 +4002,26 @@ public InlineElement getDescription() {
             return text(description);
         }
     }
+
+    /** The read mode of chain branch. */
+    public enum ChainBranchReadMode {
+        DEFAULT("default", "Read as common table."),
+        CHAIN_READ("chain_read", "Read as chain table.");
+
+        private final String value;
+        private final String description;
+
+        ChainBranchReadMode(String value, String description) {
+            this.value = value;
+            this.description = description;
+        }
+
+        public String getValue() {
+            return value;
+        }
+
+        public InlineElement getDescription() {
+            return text(description);
+        }
+    }
 }

paimon-common/src/main/java/org/apache/paimon/fs/ExternalPathProvider.java

@@ -48,4 +48,8 @@ public Path getNextExternalDataPath(String fileName) {
         }
         return new Path(new Path(externalTablePaths.get(position), relativeBucketPath), fileName);
     }
+
+    public ExternalPathProvider withBucketPath(Path relativeBucketPath) {
+        return new ExternalPathProvider(externalTablePaths, relativeBucketPath);
+    }
 }

paimon-common/src/main/java/org/apache/paimon/utils/InternalRowPartitionComputer.java

@@ -28,6 +28,7 @@
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.types.VarCharType;
 
+import java.util.ArrayList;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -142,4 +143,13 @@ public static String partToSimpleString(
         String result = builder.toString();
         return result.substring(0, Math.min(result.length(), maxLength));
     }
+
+    public List<String> generateOrderPartValues(InternalRow in) {
+        LinkedHashMap<String, String> partSpec = generatePartValues(in);
+        List<String> partValues = new ArrayList<>();
+        for (String columnName : partitionColumns) {
+            partValues.add(partSpec.get(columnName));
+        }
+        return partValues;
+    }
 }

paimon-common/src/main/java/org/apache/paimon/utils/ListUtils.java

@@ -19,7 +19,9 @@
 package org.apache.paimon.utils;
 
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.ThreadLocalRandom;
 
 import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -36,4 +38,19 @@ public static <T> T pickRandomly(List<T> list) {
     public static <T> boolean isNullOrEmpty(Collection<T> list) {
         return list == null || list.isEmpty();
     }
+
+    public static Map<String, String> convertListsToMap(List<String> keys, List<String> values) {
+        if (keys.size() != values.size()) {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "keys and values size must be equal,"
+                                    + " but got keys %s and values %s",
+                            keys, values));
+        }
+        Map<String, String> result = new HashMap<>();
+        for (int i = 0; i < keys.size(); i++) {
+            result.put(keys.get(i), values.get(i));
+        }
+        return result;
+    }
 }

paimon-common/src/main/java/org/apache/paimon/utils/RowDataToObjectArrayConverter.java

@@ -18,11 +18,16 @@
 
 package org.apache.paimon.utils;
 
+import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.data.GenericRow;
 import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
 import org.apache.paimon.types.RowType;
 
 import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.stream.IntStream;
 
 /** Convert {@link InternalRow} to object array. */
@@ -63,4 +68,36 @@ public Object[] convert(InternalRow rowData) {
         }
         return result;
     }
+
+    public Predicate createEqualPredicate(BinaryRow binaryRow) {
+        PredicateBuilder builder = new PredicateBuilder(rowType);
+        List<Predicate> fieldPredicates = new ArrayList<>();
+        Object[] partitionObjects = convert(binaryRow);
+        for (int i = 0; i < getArity(); i++) {
+            Object o = partitionObjects[i];
+            fieldPredicates.add(builder.equal(i, o));
+        }
+        return PredicateBuilder.and(fieldPredicates);
+    }
+
+    public Predicate createLessThanPredicate(BinaryRow binaryRow, boolean includeEqual) {
+        PredicateBuilder builder = new PredicateBuilder(rowType);
+        List<Predicate> fieldPredicates = new ArrayList<>();
+        Object[] partitionObjects = convert(binaryRow);
+        for (int i = 0; i < getArity(); i++) {
+            List<Predicate> andConditions = new ArrayList<>();
+            for (int j = 0; j < i; j++) {
+                Object o = partitionObjects[j];
+                andConditions.add(builder.equal(j, o));
+            }
+            Object currentValue = partitionObjects[i];
+            if (includeEqual) {
+                andConditions.add(builder.lessOrEqual(i, currentValue));
+            } else {
+                andConditions.add(builder.lessThan(i, currentValue));
+            }
+            fieldPredicates.add(PredicateBuilder.and(andConditions));
+        }
+        return PredicateBuilder.or(fieldPredicates);
+    }
 }

paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java

@@ -39,6 +39,7 @@
 import org.apache.paimon.schema.TableSchema;
 import org.apache.paimon.table.BucketMode;
 import org.apache.paimon.table.CatalogEnvironment;
+import org.apache.paimon.table.source.ScanFactory;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.KeyComparatorSupplier;
 import org.apache.paimon.utils.UserDefinedSeqComparator;
@@ -111,7 +112,7 @@ public BucketMode bucketMode() {
 
     @Override
     public MergeFileSplitRead newRead() {
-        return new MergeFileSplitRead(
+        return ScanFactory.createMergeFileSplitRead(
                 options,
                 schema,
                 keyType,
@@ -122,14 +123,14 @@ public MergeFileSplitRead newRead() {
     }
 
     public RawFileSplitRead newBatchRawFileRead() {
-        return new RawFileSplitRead(
+        return ScanFactory.createBatchRawFileRead(
                 fileIO,
                 schemaManager,
                 schema,
                 valueType,
                 FileFormatDiscover.of(options),
                 pathFactory(),
-                options.fileIndexReadEnabled(),
+                options,
                 false);
     }
 

paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java

@@ -33,6 +33,7 @@
 import org.apache.paimon.schema.Schema;
 import org.apache.paimon.schema.SchemaChange;
 import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.SchemaValidation;
 import org.apache.paimon.schema.TableSchema;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.FormatTable;
@@ -383,7 +384,10 @@ public void createTable(Identifier identifier, Schema schema, boolean ignoreIfEx
         checkNotSystemTable(identifier, "createTable");
         validateCreateTable(schema);
         validateCustomTablePath(schema.options());
-
+        SchemaValidation.validateChainTableOptions(
+                schema.options(),
+                schema.primaryKeys() == null ? null : String.join(",", schema.primaryKeys()),
+                !schema.partitionKeys().isEmpty());
         // check db exists
         getDatabase(identifier.getDatabaseName());
 

paimon-core/src/main/java/org/apache/paimon/io/ChainReadDataFilePathFactory.java

@@ -0,0 +1,80 @@
+/*
+ * 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.paimon.io;
+
+import org.apache.paimon.fs.ExternalPathProvider;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.manifest.FileEntry;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+
+/** Factory which produces read {@link Path}s data files for chain tbl. */
+public class ChainReadDataFilePathFactory extends DataFilePathFactory {
+
+    private final Map<String, String> fileBucketPathMapping;
+
+    public ChainReadDataFilePathFactory(
+            Path parent,
+            String formatIdentifier,
+            String dataFilePrefix,
+            String changelogFilePrefix,
+            boolean fileSuffixIncludeCompression,
+            String fileCompression,
+            @Nullable ExternalPathProvider externalPathProvider,
+            Map<String, String> fileBucketPathMapping) {
+        super(
+                parent,
+                formatIdentifier,
+                dataFilePrefix,
+                changelogFilePrefix,
+                fileSuffixIncludeCompression,
+                fileCompression,
+                externalPathProvider);
+        this.fileBucketPathMapping = fileBucketPathMapping;
+    }
+
+    @Override
+    public Path newPath(String prefix) {
+        throw new UnsupportedOperationException("Please use api with relativeBucketPath");
+    }
+
+    @Override
+    public Path toPath(DataFileMeta file) {
+        return file.externalPath()
+                .map(Path::new)
+                .orElse(new Path(fileBucketPathMapping.get(file.fileName()), file.fileName()));
+    }
+
+    @Override
+    public Path toPath(FileEntry file) {
+        throw new UnsupportedOperationException("Please use api with relativeBucketPath");
+    }
+
+    @Override
+    public Path toAlignedPath(String fileName, DataFileMeta aligned) {
+        throw new UnsupportedOperationException("Please use api with relativeBucketPath");
+    }
+
+    @Override
+    public Path toAlignedPath(String fileName, FileEntry aligned) {
+        throw new UnsupportedOperationException("Please use api with relativeBucketPath");
+    }
+}