This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 3d10846a9c [core] Add file index system table (#9138)
3d10846a9c is described below
commit 3d10846a9c0b1aae48778d9d36d676d15951d2b5
Author: sanshi <[email protected]>
AuthorDate: Mon Aug 10 13:41:31 2026 +0800
[core] Add file index system table (#9138)
---
docs/docs/concepts/system-tables.mdx | 29 ++
.../apache/paimon/fileindex/FileIndexFormat.java | 58 +++
.../fileindex/FileIndexFormatFormatTest.java | 38 ++
.../paimon/table/system/FileIndexesTable.java | 444 +++++++++++++++++++++
.../paimon/table/system/SystemTableLoader.java | 2 +
.../paimon/table/system/FileIndexesTableTest.java | 273 +++++++++++++
6 files changed, 844 insertions(+)
diff --git a/docs/docs/concepts/system-tables.mdx
b/docs/docs/concepts/system-tables.mdx
index d70bc6c39b..67fb73a541 100644
--- a/docs/docs/concepts/system-tables.mdx
+++ b/docs/docs/concepts/system-tables.mdx
@@ -285,6 +285,35 @@ SELECT * FROM my_table$files /*+
OPTIONS('scan.snapshot-id'='1') */;
*/
```
+### File Indexes Table
+
+You can query the file indexes of every data file in a specific snapshot
through the
+`file_indexes` table. Each row represents one index type for one column in one
data file.
+Multiple rows can therefore refer to the same index container.
+
+```sql
+SELECT * FROM my_table$file_indexes;
+
+/*
++-----------+--------+--------------------------------+--------------------+--------------+-----------+-------------+--------------+--------------+--------------------------------+--------------------+----------------------------+----------+
+| partition | bucket | file_path | file_size_in_bytes |
record_count | schema_id | column_name | index_type | storage_type |
index_file_path | index_size_in_bytes | index_container_size_in_bytes |
is_empty |
++-----------+--------+--------------------------------+--------------------+--------------+-----------+-------------+--------------+--------------+--------------------------------+--------------------+----------------------------+----------+
+| {1} | 0 | data-8f64af95-29cc-4342-adc... | 593 |
2 | 0 | id | bitmap | EMBEDDED |
<NULL> | 12 | 48 |
false |
+| {1} | 0 | data-8f64af95-29cc-4342-adc... | 593 |
2 | 0 | id | bloom-filter | EMBEDDED |
<NULL> | 16 | 48 |
false |
+| {2} | 0 | data-8b369068-0d37-4011-aa5... | 593 |
2 | 0 | id | bitmap | FILE |
data-8b369068-0d37-4011-aa5... | 12 |
48 | false |
+| {2} | 0 | data-8b369068-0d37-4011-aa5... | 593 |
2 | 0 | id | bloom-filter | FILE |
data-8b369068-0d37-4011-aa5... | 16 |
48 | false |
++-----------+--------+--------------------------------+--------------------+--------------+-----------+-------------+--------------+--------------+--------------------------------+--------------------+----------------------------+----------+
+4 rows in set
+*/
+```
+
+The system table reads only file index headers and does not load index
payloads. It lists indexes
+that physically exist in the selected snapshot; data files without file
indexes do not produce
+rows.
+
+This table is different from `table_indexes`, which lists independently
managed index files from
+the snapshot's index manifest, such as deletion vectors and global indexes.
+
### File Key Ranges Table
You can query the key ranges and file location of each data file through the
file key ranges table.
diff --git
a/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexFormat.java
b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexFormat.java
index 7a132673dd..e10a95c50c 100644
---
a/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexFormat.java
+++
b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexFormat.java
@@ -36,9 +36,11 @@ import java.io.DataOutputStream;
import java.io.EOFException;
import java.io.IOException;
import java.io.OutputStream;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
@@ -123,6 +125,43 @@ public final class FileIndexFormat {
return new Reader(inputStream, fileRowType);
}
+ /** Creates a reader for accessing header metadata without reading index
payloads. */
+ public static Reader createMetadataReader(SeekableInputStream inputStream)
{
+ return new Reader(inputStream, RowType.builder().build());
+ }
+
+ /** Metadata of one column index stored in a file index container. */
+ public static class FileIndexMeta {
+
+ private final String columnName;
+ private final String indexType;
+ private final int sizeInBytes;
+ private final boolean empty;
+
+ private FileIndexMeta(String columnName, String indexType, int
sizeInBytes, boolean empty) {
+ this.columnName = columnName;
+ this.indexType = indexType;
+ this.sizeInBytes = sizeInBytes;
+ this.empty = empty;
+ }
+
+ public String columnName() {
+ return columnName;
+ }
+
+ public String indexType() {
+ return indexType;
+ }
+
+ public int sizeInBytes() {
+ return sizeInBytes;
+ }
+
+ public boolean empty() {
+ return empty;
+ }
+ }
+
/** Writer for file index file. */
public static class Writer implements Closeable {
@@ -301,6 +340,25 @@ public final class FileIndexFormat {
.orElse(Collections.emptySet());
}
+ /** Returns the index metadata parsed from the header without reading
index payloads. */
+ public List<FileIndexMeta> indexMetas() {
+ List<FileIndexMeta> metas = new ArrayList<>();
+ for (Map.Entry<String, Map<String, Pair<Integer, Integer>>>
columnEntry :
+ header.entrySet()) {
+ for (Map.Entry<String, Pair<Integer, Integer>> indexEntry :
+ columnEntry.getValue().entrySet()) {
+ Pair<Integer, Integer> startAndLength =
indexEntry.getValue();
+ metas.add(
+ new FileIndexMeta(
+ columnEntry.getKey(),
+ indexEntry.getKey(),
+ startAndLength.getRight(),
+ startAndLength.getLeft() ==
EMPTY_INDEX_FLAG));
+ }
+ }
+ return Collections.unmodifiableList(metas);
+ }
+
private FileIndexReader getFileIndexReader(
String columnName, String indexType, Pair<Integer, Integer>
startAndLength) {
if (startAndLength.getLeft() == EMPTY_INDEX_FLAG) {
diff --git
a/paimon-common/src/test/java/org/apache/paimon/fileindex/FileIndexFormatFormatTest.java
b/paimon-common/src/test/java/org/apache/paimon/fileindex/FileIndexFormatFormatTest.java
index d9828e75a1..0f5f6e299e 100644
---
a/paimon-common/src/test/java/org/apache/paimon/fileindex/FileIndexFormatFormatTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/fileindex/FileIndexFormatFormatTest.java
@@ -31,11 +31,14 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import java.util.Random;
import static org.apache.paimon.utils.RandomUtil.randomBytes;
import static org.apache.paimon.utils.RandomUtil.randomString;
+import static org.assertj.core.api.Assertions.tuple;
/** Test for {@link FileIndexFormat}. */
public class FileIndexFormatFormatTest {
@@ -105,4 +108,39 @@ public class FileIndexFormatFormatTest {
Assertions.assertThat(new ArrayList<>(fileIndexFormatList).get(0))
.isEqualTo(EmptyFileIndexReader.INSTANCE);
}
+
+ @Test
+ public void testIndexMetas() throws IOException {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ Map<String, Map<String, byte[]>> indexes = new LinkedHashMap<>();
+ indexes.computeIfAbsent("user_id", key -> new LinkedHashMap<>())
+ .put("bitmap", new byte[] {1, 2, 3});
+ indexes.computeIfAbsent("user_id", key -> new LinkedHashMap<>())
+ .put("bloom-filter", new byte[] {4, 5});
+ indexes.computeIfAbsent("region", key -> new
LinkedHashMap<>()).put("bitmap", null);
+
+ try (FileIndexFormat.Writer writer =
FileIndexFormat.createWriter(baos)) {
+ writer.writeColumnIndexes(indexes);
+ }
+
+ List<FileIndexFormat.FileIndexMeta> metas;
+ try (FileIndexFormat.Reader reader =
+ FileIndexFormat.createMetadataReader(
+ new ByteArraySeekableStream(baos.toByteArray()))) {
+ metas = reader.indexMetas();
+ }
+
+ Assertions.assertThat(metas)
+ .extracting(
+ FileIndexFormat.FileIndexMeta::columnName,
+ FileIndexFormat.FileIndexMeta::indexType,
+ FileIndexFormat.FileIndexMeta::sizeInBytes,
+ FileIndexFormat.FileIndexMeta::empty)
+ .containsExactlyInAnyOrder(
+ tuple("user_id", "bitmap", 3, false),
+ tuple("user_id", "bloom-filter", 2, false),
+ tuple("region", "bitmap", 0, true));
+ Assertions.assertThatThrownBy(() -> metas.clear())
+ .isInstanceOf(UnsupportedOperationException.class);
+ }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/system/FileIndexesTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/system/FileIndexesTable.java
new file mode 100644
index 0000000000..f79419c449
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/table/system/FileIndexesTable.java
@@ -0,0 +1,444 @@
+/*
+ * 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.table.system;
+
+import org.apache.paimon.casting.CastExecutor;
+import org.apache.paimon.casting.CastExecutors;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.fileindex.FileIndexFormat;
+import org.apache.paimon.fileindex.FileIndexFormat.FileIndexMeta;
+import org.apache.paimon.fs.ByteArraySeekableStream;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.predicate.CompoundPredicate;
+import org.apache.paimon.predicate.LeafPredicate;
+import org.apache.paimon.predicate.LeafPredicateExtractor;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.ReadonlyTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.InnerTableRead;
+import org.apache.paimon.table.source.InnerTableScan;
+import org.apache.paimon.table.source.ReadOnceTableScan;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableRead;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.types.BigIntType;
+import org.apache.paimon.types.BooleanType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.DataFilePathFactories;
+import org.apache.paimon.utils.IteratorRecordReader;
+import org.apache.paimon.utils.PartitionPredicateHelper;
+import org.apache.paimon.utils.ProjectedRow;
+import org.apache.paimon.utils.SerializationUtils;
+
+import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.catalog.Identifier.SYSTEM_TABLE_SPLITTER;
+
+/** A {@link Table} for showing file indexes of data files in a snapshot. */
+public class FileIndexesTable implements ReadonlyTable {
+
+ private static final long serialVersionUID = 1L;
+
+ public static final String FILE_INDEXES = "file_indexes";
+
+ private static final String EMBEDDED = "EMBEDDED";
+ private static final String FILE = "FILE";
+
+ public static final RowType TABLE_TYPE =
+ new RowType(
+ Arrays.asList(
+ new DataField(0, "partition",
SerializationUtils.newStringType(true)),
+ new DataField(1, "bucket", new IntType(false)),
+ new DataField(2, "file_path",
SerializationUtils.newStringType(false)),
+ new DataField(3, "file_size_in_bytes", new
BigIntType(false)),
+ new DataField(4, "record_count", new
BigIntType(false)),
+ new DataField(5, "schema_id", new
BigIntType(false)),
+ new DataField(
+ 6, "column_name",
SerializationUtils.newStringType(false)),
+ new DataField(7, "index_type",
SerializationUtils.newStringType(false)),
+ new DataField(
+ 8, "storage_type",
SerializationUtils.newStringType(false)),
+ new DataField(
+ 9, "index_file_path",
SerializationUtils.newStringType(true)),
+ new DataField(10, "index_size_in_bytes", new
BigIntType(false)),
+ new DataField(
+ 11, "index_container_size_in_bytes", new
BigIntType(false)),
+ new DataField(12, "is_empty", new
BooleanType(false))));
+
+ private final FileStoreTable storeTable;
+
+ public FileIndexesTable(FileStoreTable storeTable) {
+ this.storeTable = storeTable;
+ }
+
+ @Override
+ public String name() {
+ return storeTable.name() + SYSTEM_TABLE_SPLITTER + FILE_INDEXES;
+ }
+
+ @Override
+ public RowType rowType() {
+ return TABLE_TYPE;
+ }
+
+ @Override
+ public List<String> primaryKeys() {
+ return Arrays.asList("file_path", "column_name", "index_type");
+ }
+
+ @Override
+ public FileIO fileIO() {
+ return storeTable.fileIO();
+ }
+
+ @Override
+ public InnerTableScan newScan() {
+ return new FileIndexesScan(storeTable);
+ }
+
+ @Override
+ public InnerTableRead newRead() {
+ return new FileIndexesRead(storeTable);
+ }
+
+ @Override
+ public Table copy(Map<String, String> dynamicOptions) {
+ return new FileIndexesTable(storeTable.copy(dynamicOptions));
+ }
+
+ private static class FileIndexesScan extends ReadOnceTableScan {
+
+ @Nullable private LeafPredicate partitionPredicate;
+ @Nullable private LeafPredicate bucketPredicate;
+
+ private final FileStoreTable storeTable;
+
+ private FileIndexesScan(FileStoreTable storeTable) {
+ this.storeTable = storeTable;
+ }
+
+ @Override
+ public InnerTableScan withFilter(Predicate pushdown) {
+ if (pushdown == null) {
+ return this;
+ }
+
+ Map<String, LeafPredicate> leafPredicates =
+ pushdown.visit(LeafPredicateExtractor.INSTANCE);
+ partitionPredicate = leafPredicates.get("partition");
+ bucketPredicate = leafPredicates.get("bucket");
+ return this;
+ }
+
+ @Override
+ public Plan innerPlan() {
+ SnapshotReader snapshotReader = storeTable.newSnapshotReader();
+ boolean hasResults =
+ PartitionPredicateHelper.applyPartitionFilter(
+ snapshotReader,
+ partitionPredicate,
+ storeTable.partitionKeys(),
+ storeTable.schema().logicalPartitionType());
+ if (!hasResults) {
+ return Collections::emptyList;
+ }
+
+ return () ->
+ snapshotReader.partitions().stream()
+ .map(
+ partition ->
+ // Keep file inventory planning
aligned with FilesTable.
+ new FilesTable.FilesSplit(
+ partition,
bucketPredicate, null))
+ .collect(Collectors.toList());
+ }
+ }
+
+ private static class FileIndexesRead implements InnerTableRead {
+
+ private static final Set<String> SCAN_PUSHDOWN_FIELDS =
+ Collections.unmodifiableSet(new
HashSet<>(Arrays.asList("partition", "bucket")));
+ private static final Set<String> FILE_PUSHDOWN_FIELDS =
Collections.singleton("file_path");
+
+ private final FileStoreTable storeTable;
+ private final DataFilePathFactories pathFactories;
+
+ @Nullable private Predicate filePredicate;
+ @Nullable private Predicate predicate;
+ @Nullable private RowType readType;
+
+ private FileIndexesRead(FileStoreTable storeTable) {
+ this.storeTable = storeTable;
+ this.pathFactories = new
DataFilePathFactories(storeTable.store().pathFactory());
+ }
+
+ @Override
+ public InnerTableRead withFilter(Predicate predicate) {
+ List<Predicate> remaining =
+ PredicateBuilder.splitAnd(predicate).stream()
+ .filter(
+ p ->
+ !(p instanceof LeafPredicate)
+ || !onlyContainsFields(p,
SCAN_PUSHDOWN_FIELDS))
+ .collect(Collectors.toList());
+ List<Predicate> filePredicates =
+ remaining.stream()
+ .filter(p -> onlyContainsFields(p,
FILE_PUSHDOWN_FIELDS))
+ .collect(Collectors.toList());
+ remaining.removeAll(filePredicates);
+
+ this.filePredicate =
+ filePredicates.isEmpty() ? null :
PredicateBuilder.and(filePredicates);
+ this.predicate = remaining.isEmpty() ? null :
PredicateBuilder.and(remaining);
+ return this;
+ }
+
+ private static boolean onlyContainsFields(Predicate predicate,
Set<String> fields) {
+ if (predicate instanceof CompoundPredicate) {
+ return ((CompoundPredicate) predicate)
+ .children().stream().allMatch(p ->
onlyContainsFields(p, fields));
+ }
+ return fields.containsAll(((LeafPredicate)
predicate).fieldNames());
+ }
+
+ @Override
+ public InnerTableRead withReadType(RowType readType) {
+ this.readType = readType;
+ return this;
+ }
+
+ @Override
+ public TableRead withIOManager(IOManager ioManager) {
+ return this;
+ }
+
+ @Override
+ public RecordReader<InternalRow> createReader(Split split) {
+ if (!(split instanceof FilesTable.FilesSplit)) {
+ throw new IllegalArgumentException("Unsupported split: " +
split.getClass());
+ }
+
+ List<Split> dataSplits = ((FilesTable.FilesSplit)
split).splits(storeTable);
+ if (dataSplits.isEmpty()) {
+ return new IteratorRecordReader<>(Collections.emptyIterator());
+ }
+
+ @SuppressWarnings("unchecked")
+ CastExecutor<InternalRow, BinaryString> partitionCastExecutor =
+ (CastExecutor<InternalRow, BinaryString>)
+ CastExecutors.resolveToString(
+
storeTable.schema().logicalPartitionType());
+
+ Iterator<InternalRow> iterator = splitRows(dataSplits,
partitionCastExecutor);
+ if (predicate != null) {
+ iterator = Iterators.filter(iterator, predicate::test);
+ }
+ if (readType != null) {
+ iterator =
+ Iterators.transform(
+ iterator,
+ row ->
+ ProjectedRow.from(readType,
FileIndexesTable.TABLE_TYPE)
+ .replaceRow(row));
+ }
+ return new IteratorRecordReader<>(iterator);
+ }
+
+ private Iterator<InternalRow> splitRows(
+ List<Split> dataSplits,
+ CastExecutor<InternalRow, BinaryString> partitionCastExecutor)
{
+ Iterator<Iterator<InternalRow>> splitRows =
+ Iterators.transform(
+ dataSplits.iterator(),
+ split -> fileRows((DataSplit) split,
partitionCastExecutor));
+ return Iterators.concat(splitRows);
+ }
+
+ private Iterator<InternalRow> fileRows(
+ DataSplit dataSplit,
+ CastExecutor<InternalRow, BinaryString> partitionCastExecutor)
{
+ DataFilePathFactory dataFilePathFactory =
+ pathFactories.get(dataSplit.partition(),
dataSplit.bucket());
+ Iterator<Iterator<InternalRow>> fileRows =
+ Iterators.transform(
+ dataSplit.dataFiles().iterator(),
+ file -> {
+ BinaryString filePath = filePath(dataSplit,
file);
+ if (filePredicate != null &&
!testFilePath(filePath)) {
+ return Collections.emptyIterator();
+ }
+ return indexRows(
+ dataSplit,
+ file,
+ filePath,
+ dataFilePathFactory,
+ partitionCastExecutor)
+ .iterator();
+ });
+ return Iterators.concat(fileRows);
+ }
+
+ private boolean testFilePath(BinaryString filePath) {
+ GenericRow row = new GenericRow(TABLE_TYPE.getFieldCount());
+ row.setField(2, filePath);
+ return filePredicate.test(row);
+ }
+
+ private List<InternalRow> indexRows(
+ DataSplit dataSplit,
+ DataFileMeta file,
+ BinaryString filePath,
+ DataFilePathFactory dataFilePathFactory,
+ CastExecutor<InternalRow, BinaryString> partitionCastExecutor)
{
+ byte[] embeddedIndex = file.embeddedIndex();
+ if (embeddedIndex != null) {
+ try (FileIndexFormat.Reader reader =
+ FileIndexFormat.createMetadataReader(
+ new ByteArraySeekableStream(embeddedIndex))) {
+ return toRows(
+ dataSplit,
+ file,
+ filePath,
+ partitionCastExecutor,
+ reader.indexMetas(),
+ EMBEDDED,
+ null,
+ embeddedIndex.length);
+ } catch (IOException e) {
+ throw new UncheckedIOException(
+ "Failed to read file index metadata from " +
filePath + ".", e);
+ } catch (RuntimeException e) {
+ throw fileIndexReadException(filePath.toString(), e);
+ }
+ }
+
+ List<String> indexFiles =
+ file.extraFiles().stream()
+ .filter(name ->
name.endsWith(DataFilePathFactory.INDEX_PATH_SUFFIX))
+ .collect(Collectors.toList());
+ if (indexFiles.isEmpty()) {
+ return Collections.emptyList();
+ }
+ if (indexFiles.size() > 1) {
+ throw new IllegalStateException(
+ "Found more than one file index for data file "
+ + file.fileName()
+ + ": "
+ + String.join(", ", indexFiles));
+ }
+
+ Path indexPath =
dataFilePathFactory.toAlignedPath(indexFiles.get(0), file);
+ try {
+ long containerSize =
storeTable.fileIO().getFileStatus(indexPath).getLen();
+ try (FileIndexFormat.Reader reader =
+ FileIndexFormat.createMetadataReader(
+
storeTable.fileIO().newInputStream(indexPath))) {
+ return toRows(
+ dataSplit,
+ file,
+ filePath,
+ partitionCastExecutor,
+ reader.indexMetas(),
+ FILE,
+ BinaryString.fromString(indexPath.toString()),
+ containerSize);
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(
+ "Failed to read file index metadata from " + indexPath
+ ".", e);
+ } catch (RuntimeException e) {
+ throw fileIndexReadException(indexPath.toString(), e);
+ }
+ }
+
+ private static RuntimeException fileIndexReadException(
+ String indexLocation, RuntimeException exception) {
+ String message = "Failed to read file index metadata from " +
indexLocation + ".";
+ if (exception.getCause() instanceof IOException) {
+ return new UncheckedIOException(message, (IOException)
exception.getCause());
+ }
+ return new RuntimeException(message, exception);
+ }
+
+ private static List<InternalRow> toRows(
+ DataSplit dataSplit,
+ DataFileMeta file,
+ BinaryString filePath,
+ CastExecutor<InternalRow, BinaryString> partitionCastExecutor,
+ List<FileIndexMeta> indexMetas,
+ String storageType,
+ @Nullable BinaryString indexFilePath,
+ long containerSize) {
+ BinaryString partition =
+ dataSplit.partition() == null
+ ? null
+ :
partitionCastExecutor.cast(dataSplit.partition());
+ List<InternalRow> rows = new ArrayList<>(indexMetas.size());
+ for (FileIndexMeta indexMeta : indexMetas) {
+ rows.add(
+ GenericRow.of(
+ partition,
+ dataSplit.bucket(),
+ filePath,
+ file.fileSize(),
+ file.rowCount(),
+ file.schemaId(),
+
BinaryString.fromString(indexMeta.columnName()),
+ BinaryString.fromString(indexMeta.indexType()),
+ BinaryString.fromString(storageType),
+ indexFilePath,
+ (long) indexMeta.sizeInBytes(),
+ containerSize,
+ indexMeta.empty()));
+ }
+ return rows;
+ }
+
+ private static BinaryString filePath(DataSplit dataSplit, DataFileMeta
file) {
+ return BinaryString.fromString(
+ file.externalPath().orElse(dataSplit.bucketPath() + "/" +
file.fileName()));
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/system/SystemTableLoader.java
b/paimon-core/src/main/java/org/apache/paimon/table/system/SystemTableLoader.java
index 1f0727c4f5..3bc083f3e2 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/system/SystemTableLoader.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/system/SystemTableLoader.java
@@ -49,6 +49,7 @@ import static
org.apache.paimon.table.system.BranchesTable.BRANCHES;
import static org.apache.paimon.table.system.BucketsTable.BUCKETS;
import static
org.apache.paimon.table.system.CatalogOptionsTable.CATALOG_OPTIONS;
import static org.apache.paimon.table.system.ConsumersTable.CONSUMERS;
+import static org.apache.paimon.table.system.FileIndexesTable.FILE_INDEXES;
import static
org.apache.paimon.table.system.FileKeyRangesTable.FILE_KEY_RANGES;
import static org.apache.paimon.table.system.FilesTable.FILES;
import static org.apache.paimon.table.system.ManifestsTable.MANIFESTS;
@@ -75,6 +76,7 @@ public class SystemTableLoader {
.put(BUCKETS, BucketsTable::new)
.put(AUDIT_LOG, AuditLogTable::new)
.put(FILES, FilesTable::new)
+ .put(FILE_INDEXES, FileIndexesTable::new)
.put(FILE_KEY_RANGES, FileKeyRangesTable::new)
.put(TAGS, TagsTable::new)
.put(BRANCHES, BranchesTable::new)
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/system/FileIndexesTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/system/FileIndexesTableTest.java
new file mode 100644
index 0000000000..a3a90d1b45
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/system/FileIndexesTableTest.java
@@ -0,0 +1,273 @@
+/*
+ * 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.table.system;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.TableTestBase;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.UncheckedIOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.catalog.Identifier.SYSTEM_TABLE_SPLITTER;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link FileIndexesTable}. */
+public class FileIndexesTableTest extends TableTestBase {
+
+ @Test
+ public void testEmbeddedFileIndexes() throws Exception {
+ FileIndexesTable fileIndexesTable = createTable("EmbeddedIndexes", "1
MB");
+
+ List<InternalRow> rows = read(fileIndexesTable);
+ assertIndexRows(rows, "EMBEDDED");
+ assertThat(rows).allMatch(row -> row.isNullAt(9));
+ assertThat(rows)
+ .allMatch(
+ row ->
+ row.getLong(11) > row.getLong(10)
+ && row.getLong(11) > 0
+ && !row.getBoolean(12));
+ }
+
+ @Test
+ public void testExternalFileIndexes() throws Exception {
+ FileIndexesTable fileIndexesTable = createTable("ExternalIndexes", "1
B");
+
+ List<InternalRow> rows = read(fileIndexesTable);
+ assertIndexRows(rows, "FILE");
+ assertThat(rows)
+ .allMatch(
+ row ->
+ row.getString(9).toString().endsWith(".index")
+ && row.getLong(11) > row.getLong(10)
+ && row.getLong(11) > 0
+ && !row.getBoolean(12));
+ assertThat(
+ rows.stream()
+ .map(row -> row.getString(9).toString())
+ .distinct()
+ .collect(Collectors.toList()))
+ .hasSize(1);
+ }
+
+ @Test
+ public void testLazyReadExternalFileIndexes() throws Exception {
+ String tableName = "LazyExternalIndexes";
+ FileIndexesTable fileIndexesTable = createTable(tableName, "1 B",
true);
+ FileStoreTable dataTable = (FileStoreTable)
catalog.getTable(identifier(tableName));
+
+ ReadBuilder readBuilder = fileIndexesTable.newReadBuilder();
+ List<Split> systemSplits = readBuilder.newScan().plan().splits();
+ assertThat(systemSplits).hasSize(1);
+
+ List<DataSplit> dataSplits =
+ ((FilesTable.FilesSplit) systemSplits.get(0))
+ .splits(dataTable).stream()
+ .map(DataSplit.class::cast)
+ .collect(Collectors.toList());
+ List<DataFileMeta> dataFiles =
+ dataSplits.stream()
+ .flatMap(split -> split.dataFiles().stream())
+ .collect(Collectors.toList());
+ assertThat(dataFiles).hasSize(2);
+
+ DataSplit firstSplit = dataSplits.get(0);
+ DataFileMeta secondFile = dataFiles.get(1);
+ String secondIndexFile =
+ secondFile.extraFiles().stream()
+ .filter(name ->
name.endsWith(DataFilePathFactory.INDEX_PATH_SUFFIX))
+ .findFirst()
+ .orElseThrow(AssertionError::new);
+ Path secondIndexPath =
+ dataTable
+ .store()
+ .pathFactory()
+ .createDataFilePathFactory(firstSplit.partition(),
firstSplit.bucket())
+ .toAlignedPath(secondIndexFile, secondFile);
+ try (PositionOutputStream output =
+ dataTable.fileIO().newOutputStream(secondIndexPath, true)) {
+ output.write(0);
+ }
+
+ try (RecordReader<InternalRow> reader =
+ readBuilder.newRead().createReader(systemSplits.get(0))) {
+ RecordReader.RecordIterator<InternalRow> batch =
reader.readBatch();
+ InternalRow first = batch.next();
+ InternalRow second = batch.next();
+ assertThat(first).isNotNull();
+ assertThat(second).isNotNull();
+ assertThat(first.getString(2)).isEqualTo(second.getString(2));
+ assertThatThrownBy(batch::next)
+ .isInstanceOf(UncheckedIOException.class)
+ .hasMessageContaining(secondIndexPath.toString());
+ batch.releaseBatch();
+ }
+ }
+
+ @Test
+ public void testFilters() throws Exception {
+ FileIndexesTable fileIndexesTable = createTable("FilteredIndexes", "1
MB");
+ PredicateBuilder builder = new
PredicateBuilder(FileIndexesTable.TABLE_TYPE);
+
+ List<InternalRow> bitmapRows =
+ readWithFilter(
+ fileIndexesTable, builder.equal(7,
BinaryString.fromString("bitmap")));
+ assertThat(bitmapRows).hasSize(1);
+
assertThat(bitmapRows.get(0).getString(7).toString()).isEqualTo("bitmap");
+
+ List<InternalRow> partitionRows =
+ readWithFilter(fileIndexesTable, builder.equal(0,
BinaryString.fromString("{1}")));
+ assertThat(partitionRows).hasSize(2);
+
+ String filePath = partitionRows.get(0).getString(2).toString();
+ List<InternalRow> fileRows =
+ readWithFilter(
+ fileIndexesTable, builder.equal(2,
BinaryString.fromString(filePath)));
+ assertThat(fileRows).hasSize(2);
+
+ assertThat(
+ readWithFilter(
+ fileIndexesTable,
+ builder.equal(2,
BinaryString.fromString(filePath + ".missing"))))
+ .isEmpty();
+
+ Predicate mixedOr =
+ PredicateBuilder.or(
+ builder.equal(0, BinaryString.fromString("{2}")),
+ builder.equal(7, BinaryString.fromString("bitmap")));
+ List<InternalRow> mixedOrRows = readWithFilter(fileIndexesTable,
mixedOr);
+ assertThat(mixedOrRows).hasSize(1);
+
assertThat(mixedOrRows.get(0).getString(7).toString()).isEqualTo("bitmap");
+ }
+
+ @Test
+ public void testTableWithoutFileIndexes() throws Exception {
+ Identifier identifier = identifier("NoIndexes");
+ catalog.createTable(
+ identifier,
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .option(CoreOptions.BUCKET.key(), "1")
+ .option(CoreOptions.BUCKET_KEY.key(), "id")
+ .build(),
+ false);
+ write((FileStoreTable) catalog.getTable(identifier), GenericRow.of(1));
+
+ FileIndexesTable fileIndexesTable =
+ (FileIndexesTable)
+ catalog.getTable(
+ identifier(
+ "NoIndexes"
+ + SYSTEM_TABLE_SPLITTER
+ +
FileIndexesTable.FILE_INDEXES));
+ assertThat(read(fileIndexesTable)).isEmpty();
+ }
+
+ private FileIndexesTable createTable(String tableName, String
inManifestThreshold)
+ throws Exception {
+ return createTable(tableName, inManifestThreshold, false);
+ }
+
+ private FileIndexesTable createTable(
+ String tableName, String inManifestThreshold, boolean
writeSeparateFiles)
+ throws Exception {
+ Identifier identifier = identifier(tableName);
+ catalog.createTable(
+ identifier,
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("pt", DataTypes.INT())
+ .partitionKeys("pt")
+ .option(CoreOptions.BUCKET.key(), "1")
+ .option(CoreOptions.BUCKET_KEY.key(), "id")
+ .option("file-index.bitmap.columns", "id")
+ .option("file-index.bloom-filter.columns", "id")
+ .option(
+
CoreOptions.FILE_INDEX_IN_MANIFEST_THRESHOLD.key(),
+ inManifestThreshold)
+ .build(),
+ false);
+ FileStoreTable table = (FileStoreTable) catalog.getTable(identifier);
+ if (writeSeparateFiles) {
+ write(table, GenericRow.of(1, 1));
+ write(table, GenericRow.of(2, 1));
+ } else {
+ write(table, GenericRow.of(1, 1), GenericRow.of(2, 1));
+ }
+
+ return (FileIndexesTable)
+ catalog.getTable(
+ identifier(
+ tableName + SYSTEM_TABLE_SPLITTER +
FileIndexesTable.FILE_INDEXES));
+ }
+
+ private static void assertIndexRows(List<InternalRow> rows, String
storageType) {
+ assertThat(rows).hasSize(2);
+ assertThat(rows).extracting(row ->
row.getString(6).toString()).containsOnly("id");
+ assertThat(rows)
+ .extracting(row -> row.getString(7).toString())
+ .containsExactlyInAnyOrder("bitmap", "bloom-filter");
+ assertThat(rows).extracting(row ->
row.getString(8).toString()).containsOnly(storageType);
+ assertThat(rows).allMatch(row ->
row.getString(0).toString().equals("{1}"));
+ assertThat(rows).allMatch(row -> row.getInt(1) == 0);
+ assertThat(rows).allMatch(row -> row.getLong(3) > 0);
+ assertThat(rows).allMatch(row -> row.getLong(4) == 2);
+ assertThat(rows).allMatch(row -> row.getLong(5) == 0);
+ assertThat(rows).allMatch(row -> row.getLong(10) > 0);
+ assertThat(
+ rows.stream()
+ .map(row -> row.getString(2).toString())
+ .distinct()
+ .collect(Collectors.toList()))
+ .hasSize(1);
+ }
+
+ private static List<InternalRow> readWithFilter(FileIndexesTable table,
Predicate predicate)
+ throws Exception {
+ ReadBuilder readBuilder = table.newReadBuilder().withFilter(predicate);
+ List<InternalRow> rows = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+ reader.forEachRemaining(rows::add);
+ }
+ return rows;
+ }
+}
