This is an automated email from the ASF dual-hosted git repository.
leaves12138 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 c48daf6580 [globalindex] Support multi-column GlobalIndex framework
(#7933)
c48daf6580 is described below
commit c48daf6580fc93be868a8c4ca1ace37606d220ee
Author: chulong89u <[email protected]>
AuthorDate: Tue Jun 16 16:15:11 2026 +0800
[globalindex] Support multi-column GlobalIndex framework (#7933)
---
.../globalindex/ConstantGlobalIndexReader.java | 132 ++++++++++++++
...tory.java => GlobalIndexMultiColumnWriter.java} | 21 ++-
.../apache/paimon/globalindex/GlobalIndexer.java | 10 +-
.../paimon/globalindex/GlobalIndexerFactory.java | 20 ++-
.../globalindex/GlobalIndexEvaluatorTest.java | 52 ++++++
.../globalindex/GlobalIndexBuilderUtils.java | 139 ++++++++++++++-
.../paimon/globalindex/GlobalIndexScanner.java | 136 ++++++++++++---
.../org/apache/paimon/index/GlobalIndexMeta.java | 46 +++++
.../paimon/manifest/IndexManifestFileHandler.java | 13 +-
.../paimon/table/source/FullTextReadImpl.java | 20 ++-
.../apache/paimon/table/source/VectorReadImpl.java | 20 ++-
.../apache/paimon/table/source/VectorScanImpl.java | 21 ++-
.../paimon/table/system/TableIndexesTable.java | 12 +-
.../globalindex/GlobalIndexBuilderUtilsTest.java | 147 ++++++++++++++++
.../dataevolution/MergeIntoUpdateChecker.java | 14 +-
.../flink/globalindex/GenericIndexTopoBuilder.java | 192 ++++++++++++++-------
.../procedure/CreateGlobalIndexProcedure.java | 51 +++++-
.../flink/procedure/DropGlobalIndexProcedure.java | 50 ++++--
.../globalindex/GenericIndexTopoBuilderTest.java | 7 +-
.../procedure/DropGlobalIndexProcedureITCase.java | 2 +-
.../MergeIntoPaimonDataEvolutionTable.scala | 8 +-
.../globalindex/DefaultGlobalIndexBuilder.java | 91 ++++++++--
.../globalindex/DefaultGlobalIndexTopoBuilder.java | 37 ++++
.../globalindex/GlobalIndexTopologyBuilder.java | 28 +++
.../procedure/CreateGlobalIndexProcedure.java | 64 +++++--
.../spark/procedure/DropGlobalIndexProcedure.java | 34 ++--
.../MergeIntoPaimonDataEvolutionTable.scala | 8 +-
27 files changed, 1180 insertions(+), 195 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/ConstantGlobalIndexReader.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/ConstantGlobalIndexReader.java
new file mode 100644
index 0000000000..fe610f9668
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/ConstantGlobalIndexReader.java
@@ -0,0 +1,132 @@
+/*
+ * 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.globalindex;
+
+import org.apache.paimon.predicate.FieldRef;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * A {@link GlobalIndexReader} that returns the same fixed result for every
scalar predicate.
+ *
+ * <p>Used to pad an index that covers a shorter row range with an all-hit
bitmap over the missing
+ * tail, so that AND-ing it with a longer-range index does not drop rows the
shorter index simply
+ * has not indexed.
+ */
+public class ConstantGlobalIndexReader implements GlobalIndexReader {
+
+ private final CompletableFuture<Optional<GlobalIndexResult>> result;
+
+ public ConstantGlobalIndexReader(GlobalIndexResult result) {
+ this.result = CompletableFuture.completedFuture(Optional.of(result));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>>
visitIsNotNull(FieldRef fieldRef) {
+ return result;
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitIsNull(FieldRef
fieldRef) {
+ return result;
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitStartsWith(
+ FieldRef fieldRef, Object literal) {
+ return result;
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitEndsWith(
+ FieldRef fieldRef, Object literal) {
+ return result;
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitContains(
+ FieldRef fieldRef, Object literal) {
+ return result;
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitLike(
+ FieldRef fieldRef, Object literal) {
+ return result;
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitLessThan(
+ FieldRef fieldRef, Object literal) {
+ return result;
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitGreaterOrEqual(
+ FieldRef fieldRef, Object literal) {
+ return result;
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitNotEqual(
+ FieldRef fieldRef, Object literal) {
+ return result;
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitLessOrEqual(
+ FieldRef fieldRef, Object literal) {
+ return result;
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitEqual(
+ FieldRef fieldRef, Object literal) {
+ return result;
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitGreaterThan(
+ FieldRef fieldRef, Object literal) {
+ return result;
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitIn(
+ FieldRef fieldRef, List<Object> literals) {
+ return result;
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitNotIn(
+ FieldRef fieldRef, List<Object> literals) {
+ return result;
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitBetween(
+ FieldRef fieldRef, Object from, Object to) {
+ return result;
+ }
+
+ @Override
+ public void close() {}
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactory.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexMultiColumnWriter.java
similarity index 52%
copy from
paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactory.java
copy to
paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexMultiColumnWriter.java
index 6eabb6d253..58a847b64c 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactory.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexMultiColumnWriter.java
@@ -18,14 +18,21 @@
package org.apache.paimon.globalindex;
-import org.apache.paimon.fileindex.FileIndexer;
-import org.apache.paimon.options.Options;
-import org.apache.paimon.types.DataField;
+import org.apache.paimon.data.InternalRow;
-/** File index factory to construct {@link FileIndexer}. */
-public interface GlobalIndexerFactory {
+import javax.annotation.Nullable;
- String identifier();
+/** Index writer for global index that accepts multiple column values per row.
*/
+public interface GlobalIndexMultiColumnWriter extends GlobalIndexWriter {
- GlobalIndexer create(DataField dataField, Options options);
+ /**
+ * Write one record's indexed columns at the given relative row id.
+ *
+ * @param rowId the record's row id relative to the current shard (0 to
rowCnt - 1); a null row
+ * still advances the row id without indexing a value
+ * @param row a projected row containing only the indexed columns, whose
layout matches the
+ * fields order passed to {@link
GlobalIndexerFactory#create(java.util.List,
+ * org.apache.paimon.options.Options)}
+ */
+ void write(long rowId, @Nullable InternalRow row);
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexer.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexer.java
index 74d223a604..5eadf0597f 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexer.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexer.java
@@ -37,8 +37,14 @@ public interface GlobalIndexer {
List<GlobalIndexIOMeta> files,
ExecutorService executor);
- static GlobalIndexer create(String type, DataField dataField, Options
options) {
+ static GlobalIndexer create(String type, DataField indexField, Options
options) {
GlobalIndexerFactory globalIndexerFactory =
GlobalIndexerFactoryUtils.load(type);
- return globalIndexerFactory.create(dataField, options);
+ return globalIndexerFactory.create(indexField, options);
+ }
+
+ static GlobalIndexer create(
+ String type, DataField indexField, List<DataField> extraFields,
Options options) {
+ GlobalIndexerFactory globalIndexerFactory =
GlobalIndexerFactoryUtils.load(type);
+ return globalIndexerFactory.create(indexField, extraFields, options);
}
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactory.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactory.java
index 6eabb6d253..741f91f42a 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactory.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactory.java
@@ -22,10 +22,28 @@ import org.apache.paimon.fileindex.FileIndexer;
import org.apache.paimon.options.Options;
import org.apache.paimon.types.DataField;
+import java.util.List;
+
/** File index factory to construct {@link FileIndexer}. */
public interface GlobalIndexerFactory {
String identifier();
- GlobalIndexer create(DataField dataField, Options options);
+ GlobalIndexer create(DataField indexField, Options options);
+
+ /**
+ * Creates an indexer over a primary column plus optional extra columns.
{@code indexField} is
+ * the primary column; {@code extraFields} holds the remaining columns and
is empty for a
+ * single-column index.
+ */
+ default GlobalIndexer create(
+ DataField indexField, List<DataField> extraFields, Options
options) {
+ if (extraFields != null && !extraFields.isEmpty()) {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Index type '%s' does not support multi-column
index, got extra columns: %s",
+ identifier(), extraFields));
+ }
+ return create(indexField, options);
+ }
}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/GlobalIndexEvaluatorTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/GlobalIndexEvaluatorTest.java
index 45d542d921..9a95d89cd0 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/GlobalIndexEvaluatorTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/GlobalIndexEvaluatorTest.java
@@ -26,6 +26,7 @@ import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RoaringNavigableMap64;
import org.junit.jupiter.api.AfterEach;
@@ -489,6 +490,57 @@ class GlobalIndexEvaluatorTest {
evaluator.close();
}
+ @Test
+ void testShorterIndexPaddedToLongestRangeNotDroppedByAnd() {
+ executor = Executors.newFixedThreadPool(2);
+ RowType rowType = rowType();
+
+ // Field c (id 2) is an extra column of two multi-column indexes:
+ // - a short index covering rows [0,4] that matches the predicate at
{1,3}
+ // - a long index covering rows [0,9] that matches the predicate at
{1,3,7,8}
+ // The evaluator AND-s both readers for the leaf. Padding the short
index over its
+ // unindexed tail (5..9) with an all-hit reader keeps the long index's
tail matches.
+ GlobalIndexReader shortIndexPadded =
+ new UnionGlobalIndexReader(
+ Arrays.asList(
+ readerReturning(resultOf(1, 3)),
+ new ConstantGlobalIndexReader(
+ GlobalIndexResult.fromRange(new
Range(5, 9)))));
+ GlobalIndexReader longIndex = readerReturning(resultOf(1, 3, 7, 8));
+
+ GlobalIndexEvaluator evaluator =
+ new GlobalIndexEvaluator(
+ rowType, fieldId -> Arrays.asList(shortIndexPadded,
longIndex));
+
+ PredicateBuilder builder = new PredicateBuilder(rowType);
+ Optional<GlobalIndexResult> result =
evaluator.evaluate(builder.equal(2, 42));
+
+ assertThat(result).isPresent();
+ assertBitmapContainsExactly(result.get().results(), 1L, 3L, 7L, 8L);
+ evaluator.close();
+ }
+
+ @Test
+ void testShorterIndexWithoutPaddingDropsTailUnderAnd() {
+ executor = Executors.newFixedThreadPool(2);
+ RowType rowType = rowType();
+
+ // Same setup as above but WITHOUT padding the short index: AND drops
the tail matches
+ // {7,8}, which is exactly the bug the padding prevents.
+ GlobalIndexReader shortIndex = readerReturning(resultOf(1, 3));
+ GlobalIndexReader longIndex = readerReturning(resultOf(1, 3, 7, 8));
+
+ GlobalIndexEvaluator evaluator =
+ new GlobalIndexEvaluator(rowType, fieldId ->
Arrays.asList(shortIndex, longIndex));
+
+ PredicateBuilder builder = new PredicateBuilder(rowType);
+ Optional<GlobalIndexResult> result =
evaluator.evaluate(builder.equal(2, 42));
+
+ assertThat(result).isPresent();
+ assertBitmapContainsExactly(result.get().results(), 1L, 3L);
+ evaluator.close();
+ }
+
@Test
void testNonFieldLeafPredicateDoesNotThrow() {
executor = Executors.newFixedThreadPool(2);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtils.java
b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtils.java
index 085423efa8..39f7fb2b0e 100644
---
a/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtils.java
+++
b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtils.java
@@ -24,24 +24,84 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.index.IndexPathFactory;
+import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.DataField;
import org.apache.paimon.utils.Range;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
/** Utils for global index build. */
public class GlobalIndexBuilderUtils {
+ private static final Logger LOG =
LoggerFactory.getLogger(GlobalIndexBuilderUtils.class);
+
+ public static List<IndexFileMeta> toIndexFileMetas(
+ FileIO fileIO,
+ IndexPathFactory indexPathFactory,
+ CoreOptions options,
+ Range range,
+ int indexFieldId,
+ String indexType,
+ List<ResultEntry> entries)
+ throws IOException {
+ return toIndexFileMetas(
+ fileIO, indexPathFactory, options, range, indexFieldId, null,
indexType, entries);
+ }
+
+ /**
+ * Builds the index file metas. The first column in {@code fields} is
treated as the primary
+ * index column (e.g. the first column in {@code CREATE ... INDEX ON (a,
b, c)}) and is stored
+ * as {@code indexFieldId}; the remaining columns go into {@code
extraFieldIds}. Callers must
+ * pass {@code fields} in the intended column order.
+ */
public static List<IndexFileMeta> toIndexFileMetas(
+ FileIO fileIO,
+ IndexPathFactory indexPathFactory,
+ CoreOptions options,
+ Range range,
+ List<DataField> fields,
+ String indexType,
+ List<ResultEntry> entries)
+ throws IOException {
+ // The first column is the primary index column and is stored as
indexFieldId; the
+ // remaining columns (if any) go into extraFieldIds.
+ int indexFieldId = fields.get(0).id();
+ int[] extraFieldIds =
+ fields.size() > 1
+ ? fields.subList(1, fields.size()).stream()
+ .mapToInt(DataField::id)
+ .toArray()
+ : null;
+ return toIndexFileMetas(
+ fileIO,
+ indexPathFactory,
+ options,
+ range,
+ indexFieldId,
+ extraFieldIds,
+ indexType,
+ entries);
+ }
+
+ private static List<IndexFileMeta> toIndexFileMetas(
FileIO fileIO,
IndexPathFactory indexPathFactory,
CoreOptions options,
Range range,
int indexFieldId,
+ @Nullable int[] extraFieldIds,
String indexType,
List<ResultEntry> entries)
throws IOException {
@@ -50,7 +110,8 @@ public class GlobalIndexBuilderUtils {
String fileName = entry.fileName();
long fileSize =
fileIO.getFileSize(indexPathFactory.toPath(fileName));
GlobalIndexMeta globalIndexMeta =
- new GlobalIndexMeta(range.from, range.to, indexFieldId,
null, entry.meta());
+ new GlobalIndexMeta(
+ range.from, range.to, indexFieldId, extraFieldIds,
entry.meta());
Path externalPathDir = options.globalIndexExternalPath();
String externalPathString = null;
@@ -78,6 +139,82 @@ public class GlobalIndexBuilderUtils {
return
globalIndexer.createWriter(createGlobalIndexFileReadWrite(table));
}
+ public static GlobalIndexWriter createIndexWriter(
+ FileStoreTable table,
+ String indexType,
+ DataField indexField,
+ List<DataField> extraFields,
+ Options options)
+ throws IOException {
+ GlobalIndexer globalIndexer =
+ GlobalIndexer.create(indexType, indexField, extraFields,
options);
+ return
globalIndexer.createWriter(createGlobalIndexFileReadWrite(table));
+ }
+
+ /**
+ * Find the minimum firstRowId among files whose schema does not contain
all index columns.
+ * Files at or beyond this rowId cannot be indexed because the column was
added later via ALTER
+ * TABLE.
+ *
+ * @return the boundary rowId, or {@link Long#MAX_VALUE} if all files
contain the columns
+ */
+ public static long findMinNonIndexableRowId(
+ SchemaManager schemaManager, List<ManifestEntry> entries,
List<String> indexColumns) {
+ Map<Long, Boolean> schemaContainsColumns = new HashMap<>();
+ long minRowId = Long.MAX_VALUE;
+ long minSchemaId = -1;
+ for (ManifestEntry entry : entries) {
+ long sid = entry.file().schemaId();
+ boolean contains =
+ schemaContainsColumns.computeIfAbsent(
+ sid,
+ id ->
schemaManager.schema(id).fieldNames().containsAll(indexColumns));
+ if (!contains && entry.file().firstRowId() != null) {
+ long rowId = entry.file().nonNullFirstRowId();
+ if (rowId < minRowId) {
+ minRowId = rowId;
+ minSchemaId = sid;
+ }
+ }
+ }
+ if (minRowId != Long.MAX_VALUE) {
+ List<String> schemaFields =
schemaManager.schema(minSchemaId).fieldNames();
+ List<String> missingColumns = new ArrayList<>();
+ for (String col : indexColumns) {
+ if (!schemaFields.contains(col)) {
+ missingColumns.add(col);
+ }
+ }
+ LOG.info(
+ "Found non-indexable files: schemaId={} missing columns
{}, boundaryRowId={}.",
+ minSchemaId,
+ missingColumns,
+ minRowId);
+ }
+ return minRowId;
+ }
+
+ /** Keep only entries whose firstRowId is strictly less than the given
boundary. */
+ public static List<ManifestEntry> filterEntriesBefore(
+ List<ManifestEntry> entries, long boundaryRowId) {
+ if (boundaryRowId == Long.MAX_VALUE) {
+ return entries;
+ }
+ List<ManifestEntry> result = new ArrayList<>();
+ for (ManifestEntry entry : entries) {
+ if (entry.file().firstRowId() != null
+ && entry.file().nonNullFirstRowId() < boundaryRowId) {
+ result.add(entry);
+ }
+ }
+ LOG.info(
+ "Filtered {} files to {} indexable files (boundaryRowId={}).",
+ entries.size(),
+ result.size(),
+ boundaryRowId);
+ return result;
+ }
+
private static GlobalIndexFileReadWrite
createGlobalIndexFileReadWrite(FileStoreTable table) {
IndexPathFactory indexPathFactory =
table.store().pathFactory().globalIndexFileFactory();
return new GlobalIndexFileReadWrite(table.fileIO(), indexPathFactory);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexScanner.java
b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexScanner.java
index 975b281833..1c8bbd3dd7 100644
---
a/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexScanner.java
+++
b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexScanner.java
@@ -53,6 +53,7 @@ import java.util.stream.Collectors;
import static org.apache.paimon.CoreOptions.GLOBAL_INDEX_THREAD_NUM;
import static org.apache.paimon.predicate.PredicateVisitor.collectFieldNames;
import static
org.apache.paimon.table.source.snapshot.TimeTravelUtil.tryTravelOrLatest;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
/** Scanner for shard-based global indexes. */
@@ -74,29 +75,96 @@ public class GlobalIndexScanner implements Closeable {
GlobalIndexReadThreadPool.getExecutorService(options.get(GLOBAL_INDEX_THREAD_NUM));
this.indexPathFactory = indexPathFactory;
GlobalIndexFileReader indexFileReader = meta ->
fileIO.newInputStream(meta.filePath());
- Map<Integer, Map<String, Map<Range, List<IndexFileMeta>>>> indexMetas
= new HashMap<>();
+ Map<Integer, IndexMetaFileGroup> indexMetas = new HashMap<>();
+ Map<Integer, List<IndexMetaFileGroup>> extraIndexMetas = new
HashMap<>();
for (IndexFileMeta indexFile : indexFiles) {
GlobalIndexMeta meta = checkNotNull(indexFile.globalIndexMeta());
- int fieldId = meta.indexFieldId();
String indexType = indexFile.indexType();
- indexMetas
- .computeIfAbsent(fieldId, k -> new HashMap<>())
- .computeIfAbsent(indexType, k -> new HashMap<>())
- .computeIfAbsent(
- new Range(meta.rowRangeStart(),
meta.rowRangeEnd()),
- k -> new ArrayList<>())
- .add(indexFile);
+ Range range = new Range(meta.rowRangeStart(), meta.rowRangeEnd());
+ int indexFieldId = meta.indexFieldId();
+ List<Integer> fieldIds = meta.getIndexedFieldIds();
+ IndexMetaFileGroup group = indexMetas.get(indexFieldId);
+ if (group == null) {
+ group = new IndexMetaFileGroup(indexFieldId, fieldIds);
+ indexMetas.put(indexFieldId, group);
+ if (meta.extraFieldIds() != null) {
+ for (int extra : meta.extraFieldIds()) {
+ extraIndexMetas.computeIfAbsent(extra, k -> new
ArrayList<>()).add(group);
+ }
+ }
+ } else {
+ checkArgument(
+ group.fieldIds.equals(fieldIds),
+ "Primary field %s owns multiple indexes with different
columns %s and %s; "
+ + "a primary column can own at most one
index.",
+ indexFieldId,
+ group.fieldIds,
+ fieldIds);
+ }
+ group.addFile(indexType, range, indexFile);
}
IntFunction<Collection<GlobalIndexReader>> readersFunction =
- fieldId ->
- createReaders(
- indexFileReader,
- indexMetas.get(fieldId),
- rowType.getField(fieldId));
+ fId -> {
+ IndexMetaFileGroup group = indexMetas.get(fId);
+ if (group != null) {
+ return createReaders(indexFileReader, group, rowType,
Long.MIN_VALUE);
+ }
+ List<IndexMetaFileGroup> extraGroups =
extraIndexMetas.get(fId);
+ if (extraGroups == null || extraGroups.isEmpty()) {
+ return Collections.emptyList();
+ }
+ long maxEnd = Long.MIN_VALUE;
+ for (IndexMetaFileGroup g : extraGroups) {
+ maxEnd = Math.max(maxEnd, g.coverageEnd());
+ }
+ List<GlobalIndexReader> allReaders = new ArrayList<>();
+ for (IndexMetaFileGroup g : extraGroups) {
+ allReaders.addAll(createReaders(indexFileReader, g,
rowType, maxEnd));
+ }
+ return allReaders;
+ };
this.globalIndexEvaluator = new GlobalIndexEvaluator(rowType,
readersFunction);
}
+ /** All index files of one global index (single- or multi-column), grouped
for reading. */
+ private static class IndexMetaFileGroup {
+
+ private final int indexFieldId;
+ private final List<Integer> fieldIds;
+ private final Map<String, Map<Range, List<IndexFileMeta>>> metas = new
HashMap<>();
+ private long coverageEnd = Long.MIN_VALUE;
+
+ IndexMetaFileGroup(int indexFieldId, List<Integer> fieldIds) {
+ this.indexFieldId = indexFieldId;
+ this.fieldIds = fieldIds;
+ }
+
+ void addFile(String indexType, Range range, IndexFileMeta indexFile) {
+ coverageEnd = Math.max(coverageEnd, range.to);
+ metas.computeIfAbsent(indexType, k -> new HashMap<>())
+ .computeIfAbsent(range, k -> new ArrayList<>())
+ .add(indexFile);
+ }
+
+ /** The largest indexed rowId across all files of this index (ranges
start at 0). */
+ long coverageEnd() {
+ return coverageEnd;
+ }
+
+ /** The primary index column. */
+ DataField indexField(RowType rowType) {
+ return rowType.getField(indexFieldId);
+ }
+
+ /** The extra columns beyond the primary one; empty for a
single-column index. */
+ List<DataField> extraFields(RowType rowType) {
+ return fieldIds.subList(1, fieldIds.size()).stream()
+ .map(rowType::getField)
+ .collect(Collectors.toList());
+ }
+ }
+
public static Optional<GlobalIndexScanner> create(
FileStoreTable table, Collection<IndexFileMeta> indexFiles) {
if (indexFiles.isEmpty()) {
@@ -127,7 +195,19 @@ public class GlobalIndexScanner implements Closeable {
if (globalIndex == null) {
return false;
}
- return filterFieldIds.contains(globalIndex.indexFieldId());
+ // Collect indexes whose primary column is filtered, and
also multi-column
+ // indexes that have a filtered column as an extra (used
as a fallback).
+ if (filterFieldIds.contains(globalIndex.indexFieldId())) {
+ return true;
+ }
+ if (globalIndex.extraFieldIds() != null) {
+ for (int id : globalIndex.extraFieldIds()) {
+ if (filterFieldIds.contains(id)) {
+ return true;
+ }
+ }
+ }
+ return false;
};
List<IndexFileMeta> indexFiles =
@@ -144,22 +224,25 @@ public class GlobalIndexScanner implements Closeable {
private Collection<GlobalIndexReader> createReaders(
GlobalIndexFileReader indexFileReadWrite,
- Map<String, Map<Range, List<IndexFileMeta>>> indexMetas,
- DataField dataField) {
- if (indexMetas == null) {
- return Collections.emptyList();
- }
+ IndexMetaFileGroup group,
+ RowType rowType,
+ long padToEnd) {
+ DataField indexField = group.indexField(rowType);
+ List<DataField> extraFields = group.extraFields(rowType);
Set<GlobalIndexReader> readers = new HashSet<>();
- for (Map.Entry<String, Map<Range, List<IndexFileMeta>>> entry :
indexMetas.entrySet()) {
+ for (Map.Entry<String, Map<Range, List<IndexFileMeta>>> entry :
group.metas.entrySet()) {
String indexType = entry.getKey();
Map<Range, List<IndexFileMeta>> metas = entry.getValue();
GlobalIndexerFactory globalIndexerFactory =
GlobalIndexerFactoryUtils.load(indexType);
- GlobalIndexer globalIndexer =
globalIndexerFactory.create(dataField, options);
+ GlobalIndexer globalIndexer =
+ globalIndexerFactory.create(indexField, extraFields,
options);
+ long typeEnd = Long.MIN_VALUE;
List<CompletableFuture<GlobalIndexReader>> futures = new
ArrayList<>(metas.size());
for (Map.Entry<Range, List<IndexFileMeta>> rangeMetas :
metas.entrySet()) {
Range range = rangeMetas.getKey();
+ typeEnd = Math.max(typeEnd, range.to);
List<IndexFileMeta> indexFileMetas = rangeMetas.getValue();
List<GlobalIndexIOMeta> globalMetas =
indexFileMetas.stream()
@@ -177,10 +260,17 @@ public class GlobalIndexScanner implements Closeable {
}
CompletableFuture.allOf(futures.toArray(new
CompletableFuture[0])).join();
- List<GlobalIndexReader> unionReader = new
ArrayList<>(futures.size());
+ List<GlobalIndexReader> unionReader = new
ArrayList<>(futures.size() + 1);
for (CompletableFuture<GlobalIndexReader> future : futures) {
unionReader.add(future.join());
}
+ // Pad this index's missing tail with an all-hit reader so AND-ing
it with a
+ // longer-range index does not drop rows it has not indexed
(ranges start at 0).
+ if (padToEnd > typeEnd) {
+ unionReader.add(
+ new ConstantGlobalIndexReader(
+ GlobalIndexResult.fromRange(new Range(typeEnd
+ 1, padToEnd))));
+ }
readers.add(new UnionGlobalIndexReader(unionReader));
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/GlobalIndexMeta.java
b/paimon-core/src/main/java/org/apache/paimon/index/GlobalIndexMeta.java
index c468bbffb3..83a6224f3e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/GlobalIndexMeta.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/GlobalIndexMeta.java
@@ -27,7 +27,9 @@ import org.apache.paimon.utils.Range;
import javax.annotation.Nullable;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.List;
/** Schema for global index. */
public class GlobalIndexMeta {
@@ -87,4 +89,48 @@ public class GlobalIndexMeta {
public byte[] indexMeta() {
return indexMeta;
}
+
+ /** All indexed field ids in order: the primary {@link #indexFieldId}
followed by the rest. */
+ public List<Integer> getIndexedFieldIds() {
+ List<Integer> ids = new ArrayList<>();
+ ids.add(indexFieldId);
+ if (extraFieldIds != null) {
+ for (int id : extraFieldIds) {
+ ids.add(id);
+ }
+ }
+ return ids;
+ }
+
+ public List<DataField> getIndexedFields(RowType rowType) {
+ List<DataField> fields = new ArrayList<>();
+ for (int id : getIndexedFieldIds()) {
+ fields.add(rowType.getField(id));
+ }
+ return fields;
+ }
+
+ /** The primary index column. */
+ public DataField getIndexField(RowType rowType) {
+ return rowType.getField(indexFieldId);
+ }
+
+ /** The extra columns beyond the primary one; empty for a single-column
index. */
+ public List<DataField> getExtraFields(RowType rowType) {
+ List<DataField> fields = new ArrayList<>();
+ if (extraFieldIds != null) {
+ for (int id : extraFieldIds) {
+ fields.add(rowType.getField(id));
+ }
+ }
+ return fields;
+ }
+
+ public List<String> getIndexedFieldNames(RowType rowType) {
+ List<String> names = new ArrayList<>();
+ for (int id : getIndexedFieldIds()) {
+ names.add(rowType.getField(id).name());
+ }
+ return names;
+ }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFileHandler.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFileHandler.java
index 3621483197..f992780855 100644
---
a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFileHandler.java
+++
b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFileHandler.java
@@ -28,6 +28,7 @@ import org.apache.paimon.utils.Range;
import javax.annotation.Nullable;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -241,11 +242,13 @@ public class IndexManifestFileHandler {
GlobalIndexMeta addedMeta =
added.indexFile().globalIndexMeta();
if (addedMeta == null
|| retainedMeta.indexFieldId() !=
addedMeta.indexFieldId()
- || !Range.intersect(
- retainedMeta.rowRangeStart(),
- retainedMeta.rowRangeEnd(),
- addedMeta.rowRangeStart(),
- addedMeta.rowRangeEnd())) {
+ || (Arrays.equals(
+ retainedMeta.extraFieldIds(),
addedMeta.extraFieldIds())
+ && !Range.intersect(
+ retainedMeta.rowRangeStart(),
+ retainedMeta.rowRangeEnd(),
+ addedMeta.rowRangeStart(),
+ addedMeta.rowRangeEnd()))) {
continue;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextReadImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextReadImpl.java
index 66e509de89..0ca65cd05f 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextReadImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextReadImpl.java
@@ -78,10 +78,22 @@ public class FullTextReadImpl implements FullTextRead {
return GlobalIndexResult.createEmpty();
}
- String indexType =
splits.get(0).fullTextIndexFiles().get(0).indexType();
- GlobalIndexer globalIndexer =
- GlobalIndexerFactoryUtils.load(indexType)
- .create(textColumn,
table.coreOptions().toConfiguration());
+ IndexFileMeta firstFile = splits.get(0).fullTextIndexFiles().get(0);
+ String indexType = firstFile.indexType();
+ GlobalIndexMeta firstMeta = checkNotNull(firstFile.globalIndexMeta());
+ GlobalIndexer globalIndexer;
+ if (firstMeta.extraFieldIds() != null) {
+ globalIndexer =
+ GlobalIndexerFactoryUtils.load(indexType)
+ .create(
+ firstMeta.getIndexField(table.rowType()),
+ firstMeta.getExtraFields(table.rowType()),
+ table.coreOptions().toConfiguration());
+ } else {
+ globalIndexer =
+ GlobalIndexerFactoryUtils.load(indexType)
+ .create(textColumn,
table.coreOptions().toConfiguration());
+ }
IndexPathFactory indexPathFactory =
table.store().pathFactory().globalIndexFileFactory();
int parallelism =
table.coreOptions().toConfiguration().get(GLOBAL_INDEX_THREAD_NUM);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorReadImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/VectorReadImpl.java
index a4ef24637d..1b3601619c 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorReadImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/VectorReadImpl.java
@@ -105,10 +105,22 @@ public class VectorReadImpl implements VectorRead,
Serializable {
RoaringNavigableMap64 preFilter = preFilter(splits).orElse(null);
- String indexType = splits.get(0).vectorIndexFiles().get(0).indexType();
- GlobalIndexer globalIndexer =
- GlobalIndexerFactoryUtils.load(indexType)
- .create(vectorColumn,
table.coreOptions().toConfiguration());
+ IndexFileMeta firstFile = splits.get(0).vectorIndexFiles().get(0);
+ String indexType = firstFile.indexType();
+ GlobalIndexMeta firstMeta = checkNotNull(firstFile.globalIndexMeta());
+ GlobalIndexer globalIndexer;
+ if (firstMeta.extraFieldIds() != null) {
+ globalIndexer =
+ GlobalIndexerFactoryUtils.load(indexType)
+ .create(
+ firstMeta.getIndexField(table.rowType()),
+ firstMeta.getExtraFields(table.rowType()),
+ table.coreOptions().toConfiguration());
+ } else {
+ globalIndexer =
+ GlobalIndexerFactoryUtils.load(indexType)
+ .create(vectorColumn,
table.coreOptions().toConfiguration());
+ }
IndexPathFactory indexPathFactory =
table.store().pathFactory().globalIndexFileFactory();
int parallelism =
table.coreOptions().toConfiguration().get(GLOBAL_INDEX_THREAD_NUM);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorScanImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/VectorScanImpl.java
index d3db6dd13d..b59363a326 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorScanImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/VectorScanImpl.java
@@ -82,7 +82,18 @@ public class VectorScanImpl implements VectorScan {
return false;
}
int fieldId = globalIndex.indexFieldId();
- return vectorColumn.id() == fieldId ||
filterFieldIds.contains(fieldId);
+ if (vectorColumn.id() == fieldId ||
filterFieldIds.contains(fieldId)) {
+ return true;
+ }
+ int[] extras = globalIndex.extraFieldIds();
+ if (extras != null) {
+ for (int extra : extras) {
+ if (filterFieldIds.contains(extra)) {
+ return true;
+ }
+ }
+ }
+ return false;
};
List<IndexFileMeta> allIndexFiles =
@@ -94,7 +105,7 @@ public class VectorScanImpl implements VectorScan {
Map<Range, List<IndexFileMeta>> vectorByRange = new HashMap<>();
for (IndexFileMeta indexFile : allIndexFiles) {
GlobalIndexMeta meta = checkNotNull(indexFile.globalIndexMeta());
- if (meta.indexFieldId() == vectorColumn.id()) {
+ if (isPrimaryColumn(meta, vectorColumn.id())) {
Range range = new Range(meta.rowRangeStart(),
meta.rowRangeEnd());
vectorByRange.computeIfAbsent(range, k -> new
ArrayList<>()).add(indexFile);
}
@@ -111,7 +122,7 @@ public class VectorScanImpl implements VectorScan {
f -> {
GlobalIndexMeta globalIndex =
checkNotNull(f.globalIndexMeta());
- if (globalIndex.indexFieldId() ==
vectorColumn.id()) {
+ if (isPrimaryColumn(globalIndex,
vectorColumn.id())) {
return false;
}
return
range.hasIntersection(globalIndex.rowRange());
@@ -122,4 +133,8 @@ public class VectorScanImpl implements VectorScan {
return () -> splits;
}
+
+ private static boolean isPrimaryColumn(GlobalIndexMeta meta, int fieldId) {
+ return meta.indexFieldId() == fieldId;
+ }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/system/TableIndexesTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/system/TableIndexesTable.java
index 320257ce10..9ad88e977b 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/system/TableIndexesTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/system/TableIndexesTable.java
@@ -235,8 +235,16 @@ public class TableIndexesTable implements ReadonlyTable {
String indexFieldName = null;
if (globalMeta != null) {
try {
- indexFieldName =
logicalRowType.getField(globalMeta.indexFieldId()).name();
- } catch (RuntimeException ignored) {
+ indexFieldName =
+ String.join(",",
globalMeta.getIndexedFieldNames(logicalRowType));
+ } catch (RuntimeException e) {
+ // Indexed columns may no longer exist in the current
schema (e.g. dropped via
+ // ALTER TABLE); leave the name empty instead of failing
the listing.
+ LOG.debug(
+ "Failed to resolve indexed field names for index
file {} (primary field {}).",
+ indexManifestEntry.indexFile().fileName(),
+ globalMeta.indexFieldId(),
+ e);
}
}
return GenericRow.of(
diff --git
a/paimon-core/src/test/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtilsTest.java
b/paimon-core/src/test/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtilsTest.java
new file mode 100644
index 0000000000..67852ae925
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/globalindex/GlobalIndexBuilderUtilsTest.java
@@ -0,0 +1,147 @@
+/*
+ * 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.globalindex;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.IndexPathFactory;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.FloatType;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.VarCharType;
+import org.apache.paimon.utils.Range;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link GlobalIndexBuilderUtils}. */
+class GlobalIndexBuilderUtilsTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ private FileIO fileIO;
+ private IndexPathFactory indexPathFactory;
+ private CoreOptions coreOptions;
+
+ @BeforeEach
+ void setUp() {
+ fileIO = new LocalFileIO();
+ Path dir = new Path(tempDir.toString());
+ indexPathFactory =
+ new IndexPathFactory() {
+ @Override
+ public Path toPath(String fileName) {
+ return new Path(dir, fileName);
+ }
+
+ @Override
+ public Path newPath() {
+ return new Path(dir, UUID.randomUUID().toString());
+ }
+
+ @Override
+ public boolean isExternalPath() {
+ return false;
+ }
+ };
+ coreOptions = new CoreOptions(new Options().toMap());
+ }
+
+ // Test: 2 columns (title + vec), primary column title is indexFieldId,
rest in extraFieldIds
+ @Test
+ void testToIndexFileMetasMultiColumn() throws IOException {
+ DataField titleField = new DataField(1, "title", new
VarCharType(Integer.MAX_VALUE));
+ DataField vecField = new DataField(2, "vec", new ArrayType(new
FloatType()));
+ List<DataField> fields = Arrays.asList(titleField, vecField);
+
+ List<ResultEntry> entries = createDummyResultEntries();
+ Range range = new Range(0, 99);
+
+ List<IndexFileMeta> metas =
+ GlobalIndexBuilderUtils.toIndexFileMetas(
+ fileIO, indexPathFactory, coreOptions, range, fields,
"test-type", entries);
+
+ assertThat(metas).hasSize(1);
+ assertThat(metas.get(0).globalIndexMeta().indexFieldId()).isEqualTo(1);
+
assertThat(metas.get(0).globalIndexMeta().extraFieldIds()).isEqualTo(new int[]
{2});
+
assertThat(metas.get(0).globalIndexMeta().rowRangeStart()).isEqualTo(0);
+ assertThat(metas.get(0).globalIndexMeta().rowRangeEnd()).isEqualTo(99);
+ }
+
+ // Test: single column, extraFieldIds should be null (backward compatible
with single-column
+ // path)
+ @Test
+ void testToIndexFileMetasSingleColumn() throws IOException {
+ DataField titleField = new DataField(1, "title", new
VarCharType(Integer.MAX_VALUE));
+ List<DataField> fields = Collections.singletonList(titleField);
+
+ List<ResultEntry> entries = createDummyResultEntries();
+ Range range = new Range(0, 49);
+
+ List<IndexFileMeta> metas =
+ GlobalIndexBuilderUtils.toIndexFileMetas(
+ fileIO, indexPathFactory, coreOptions, range, fields,
"test-type", entries);
+
+ assertThat(metas).hasSize(1);
+ assertThat(metas.get(0).globalIndexMeta().indexFieldId()).isEqualTo(1);
+ assertThat(metas.get(0).globalIndexMeta().extraFieldIds()).isNull();
+ }
+
+ // Test: 3 columns (title + vec + id), primary column title is
indexFieldId, rest in
+ // extraFieldIds
+ @Test
+ void testToIndexFileMetasThreeColumns() throws IOException {
+ DataField titleField = new DataField(1, "title", new
VarCharType(Integer.MAX_VALUE));
+ DataField vecField = new DataField(2, "vec", new ArrayType(new
FloatType()));
+ DataField idField = new DataField(3, "id", new IntType());
+ List<DataField> fields = Arrays.asList(titleField, vecField, idField);
+
+ List<ResultEntry> entries = createDummyResultEntries();
+ Range range = new Range(0, 199);
+
+ List<IndexFileMeta> metas =
+ GlobalIndexBuilderUtils.toIndexFileMetas(
+ fileIO, indexPathFactory, coreOptions, range, fields,
"test-type", entries);
+
+ assertThat(metas).hasSize(1);
+ assertThat(metas.get(0).globalIndexMeta().indexFieldId()).isEqualTo(1);
+
assertThat(metas.get(0).globalIndexMeta().extraFieldIds()).isEqualTo(new int[]
{2, 3});
+ }
+
+ private List<ResultEntry> createDummyResultEntries() throws IOException {
+ String fileName = "test-index-" + UUID.randomUUID();
+ Path filePath = indexPathFactory.toPath(fileName);
+ fileIO.newOutputStream(filePath, false).close();
+ return Collections.singletonList(new ResultEntry(fileName, 100, null));
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/MergeIntoUpdateChecker.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/MergeIntoUpdateChecker.java
index 8b1122382a..bdd0c0d491 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/MergeIntoUpdateChecker.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/MergeIntoUpdateChecker.java
@@ -100,10 +100,12 @@ public class MergeIntoUpdateChecker extends
BoundedOneInputOperator<Committable,
GlobalIndexMeta globalIndexMeta =
entry.indexFile().globalIndexMeta();
if (globalIndexMeta != null) {
- String fieldName =
-
rowType.getField(globalIndexMeta.indexFieldId())
- .name();
- return
updatedColumns.contains(fieldName)
+ List<String> indexedNames =
+
globalIndexMeta.getIndexedFieldNames(rowType);
+ boolean overlaps =
+ indexedNames.stream()
+
.anyMatch(updatedColumns::contains);
+ return overlaps
&&
affectedPartitions.contains(entry.partition());
}
return false;
@@ -116,8 +118,8 @@ public class MergeIntoUpdateChecker extends
BoundedOneInputOperator<Committable,
case THROW_ERROR:
Set<String> conflictedColumns =
affectedEntries.stream()
- .map(file ->
file.indexFile().globalIndexMeta().indexFieldId())
- .map(id -> rowType.getField(id).name())
+ .map(file ->
file.indexFile().globalIndexMeta())
+ .flatMap(meta ->
meta.getIndexedFieldNames(rowType).stream())
.collect(Collectors.toSet());
throw new RuntimeException(
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilder.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilder.java
index 5896503ce0..af256da8ec 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilder.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilder.java
@@ -29,7 +29,9 @@ import org.apache.paimon.flink.sink.StoreCommitter;
import org.apache.paimon.flink.utils.BoundedOneInputOperator;
import org.apache.paimon.flink.utils.JavaTypeInfo;
import org.apache.paimon.flink.utils.StreamExecutionEnvironmentUtils;
+import org.apache.paimon.globalindex.GlobalIndexMultiColumnWriter;
import org.apache.paimon.globalindex.GlobalIndexSingletonWriter;
+import org.apache.paimon.globalindex.GlobalIndexWriter;
import org.apache.paimon.globalindex.ResultEntry;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.io.DataFileMeta;
@@ -38,7 +40,6 @@ import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.reader.RecordReader;
-import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.SpecialFields;
import org.apache.paimon.table.sink.BatchWriteBuilder;
@@ -50,6 +51,7 @@ import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.CloseableIterator;
+import org.apache.paimon.utils.ProjectedRow;
import org.apache.paimon.utils.Range;
import org.apache.flink.streaming.api.datastream.DataStream;
@@ -65,7 +67,6 @@ import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
-import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -74,6 +75,8 @@ import java.util.function.Supplier;
import java.util.stream.Collectors;
import static
org.apache.paimon.globalindex.GlobalIndexBuilderUtils.createIndexWriter;
+import static
org.apache.paimon.globalindex.GlobalIndexBuilderUtils.filterEntriesBefore;
+import static
org.apache.paimon.globalindex.GlobalIndexBuilderUtils.findMinNonIndexableRowId;
import static
org.apache.paimon.globalindex.GlobalIndexBuilderUtils.toIndexFileMetas;
import static org.apache.paimon.io.CompactIncrement.emptyIncrement;
import static org.apache.paimon.io.DataIncrement.deleteIndexIncrement;
@@ -104,6 +107,7 @@ public class GenericIndexTopoBuilder {
env,
table,
indexColumn,
+ Collections.emptyList(),
indexType,
partitionPredicate,
userOptions,
@@ -119,12 +123,54 @@ public class GenericIndexTopoBuilder {
Options userOptions,
long maxIndexedRowId)
throws Exception {
+ buildIndexAndExecute(
+ env,
+ table,
+ indexColumn,
+ Collections.emptyList(),
+ indexType,
+ partitionPredicate,
+ userOptions,
+ maxIndexedRowId);
+ }
+
+ public static void buildIndexAndExecute(
+ StreamExecutionEnvironment env,
+ FileStoreTable table,
+ String indexColumn,
+ List<String> extraColumns,
+ String indexType,
+ PartitionPredicate partitionPredicate,
+ Options userOptions)
+ throws Exception {
+ buildIndexAndExecute(
+ env,
+ table,
+ indexColumn,
+ extraColumns,
+ indexType,
+ partitionPredicate,
+ userOptions,
+ NO_MAX_INDEXED_ROW_ID);
+ }
+
+ public static void buildIndexAndExecute(
+ StreamExecutionEnvironment env,
+ FileStoreTable table,
+ String indexColumn,
+ List<String> extraColumns,
+ String indexType,
+ PartitionPredicate partitionPredicate,
+ Options userOptions,
+ long maxIndexedRowId)
+ throws Exception {
boolean hasIndexToBuild =
buildIndex(
env,
() -> new GenericGlobalIndexBuilder(table),
table,
indexColumn,
+ extraColumns,
indexType,
partitionPredicate,
userOptions,
@@ -150,12 +196,35 @@ public class GenericIndexTopoBuilder {
indexBuilderSupplier,
table,
indexColumn,
+ Collections.emptyList(),
indexType,
partitionPredicate,
userOptions,
NO_MAX_INDEXED_ROW_ID);
}
+ public static boolean buildIndex(
+ StreamExecutionEnvironment env,
+ Supplier<GenericGlobalIndexBuilder> indexBuilderSupplier,
+ FileStoreTable table,
+ String indexColumn,
+ String indexType,
+ PartitionPredicate partitionPredicate,
+ Options userOptions,
+ long maxIndexedRowId)
+ throws Exception {
+ return buildIndex(
+ env,
+ indexBuilderSupplier,
+ table,
+ indexColumn,
+ Collections.emptyList(),
+ indexType,
+ partitionPredicate,
+ userOptions,
+ maxIndexedRowId);
+ }
+
/**
* Builds a generic global index topology using a {@link
GenericGlobalIndexBuilder} supplier.
*
@@ -167,6 +236,7 @@ public class GenericIndexTopoBuilder {
Supplier<GenericGlobalIndexBuilder> indexBuilderSupplier,
FileStoreTable table,
String indexColumn,
+ List<String> extraColumns,
String indexType,
PartitionPredicate partitionPredicate,
Options userOptions,
@@ -184,6 +254,7 @@ public class GenericIndexTopoBuilder {
env,
table,
indexColumn,
+ extraColumns,
indexType,
userOptions,
entries,
@@ -204,32 +275,39 @@ public class GenericIndexTopoBuilder {
StreamExecutionEnvironment env,
FileStoreTable table,
String indexColumn,
+ List<String> extraColumns,
String indexType,
Options userOptions,
List<ManifestEntry> entries,
List<IndexManifestEntry> deletedIndexEntries,
long maxIndexedRowId)
throws Exception {
+ // The primary column followed by the extra columns, in index order.
+ List<String> indexColumns = new ArrayList<>(1 + extraColumns.size());
+ indexColumns.add(indexColumn);
+ indexColumns.addAll(extraColumns);
+
long totalRowCount = entries.stream().mapToLong(e ->
e.file().rowCount()).sum();
LOG.info(
- "Scanned {} files ({} rows) across {} partitions for {} index
on column '{}'"
+ "Scanned {} files ({} rows) across {} partitions for {} index
on columns '{}'"
+ (maxIndexedRowId >= 0 ? ", maxIndexedRowId={}." :
"."),
entries.size(),
totalRowCount,
entries.stream().map(ManifestEntry::partition).distinct().count(),
indexType,
- indexColumn,
+ indexColumns,
maxIndexedRowId);
long minNonIndexableRowId =
- findMinNonIndexableRowId(table.schemaManager(), entries,
indexColumn);
+ findMinNonIndexableRowId(table.schemaManager(), entries,
indexColumns);
entries = filterEntriesBefore(entries, minNonIndexableRowId);
RowType rowType = table.rowType();
DataField indexField = rowType.getField(indexColumn);
- // Project indexColumn + _ROW_ID so we can read the actual row ID from
data
- List<String> readColumns = new ArrayList<>();
- readColumns.add(indexColumn);
+ List<DataField> extraFields =
+
extraColumns.stream().map(rowType::getField).collect(Collectors.toList());
+ // Project indexColumns + _ROW_ID so we can read the actual row ID
from data
+ List<String> readColumns = new ArrayList<>(indexColumns);
readColumns.add(SpecialFields.ROW_ID.name());
RowType projectedRowType =
SpecialFields.rowTypeWithRowId(rowType).project(readColumns);
@@ -278,6 +356,7 @@ public class GenericIndexTopoBuilder {
table,
indexType,
indexField,
+ extraFields,
projectedRowType,
mergedOptions))
.setParallelism(parallelism);
@@ -298,49 +377,6 @@ public class GenericIndexTopoBuilder {
return true;
}
- /**
- * Find the minimum firstRowId among files whose schema does not contain
the index column. Files
- * at or beyond this rowId cannot be indexed because the column was added
later via ALTER TABLE.
- *
- * @return the boundary rowId, or {@link Long#MAX_VALUE} if all files
contain the column
- */
- static long findMinNonIndexableRowId(
- SchemaManager schemaManager, List<ManifestEntry> entries, String
indexColumn) {
- Map<Long, Boolean> schemaContainsColumn = new HashMap<>();
- long minRowId = Long.MAX_VALUE;
- for (ManifestEntry entry : entries) {
- long sid = entry.file().schemaId();
- boolean contains =
- schemaContainsColumn.computeIfAbsent(
- sid, id ->
schemaManager.schema(id).fieldNames().contains(indexColumn));
- if (!contains && entry.file().firstRowId() != null) {
- minRowId = Math.min(minRowId,
entry.file().nonNullFirstRowId());
- }
- }
- return minRowId;
- }
-
- /** Keep only entries whose firstRowId is strictly less than the given
boundary. */
- static List<ManifestEntry> filterEntriesBefore(
- List<ManifestEntry> entries, long boundaryRowId) {
- if (boundaryRowId == Long.MAX_VALUE) {
- return entries;
- }
- List<ManifestEntry> result = new ArrayList<>();
- for (ManifestEntry entry : entries) {
- if (entry.file().firstRowId() != null
- && entry.file().nonNullFirstRowId() < boundaryRowId) {
- result.add(entry);
- }
- }
- LOG.info(
- "Filtered {} files at or beyond rowId {}, {} files remain.",
- entries.size() - result.size(),
- boundaryRowId,
- result.size());
- return result;
- }
-
/**
* Compute shard tasks for a full build (no rows to skip).
*
@@ -549,24 +585,30 @@ public class GenericIndexTopoBuilder {
private final FileStoreTable table;
private final String indexType;
private final DataField indexField;
+ private final List<DataField> extraFields;
private final RowType projectedRowType;
private final Options mergedOptions;
private transient TableRead tableRead;
- private transient InternalRow.FieldGetter indexFieldGetter;
+ private transient List<DataField> indexedFields;
+ private transient InternalRow.FieldGetter[] indexFieldGetters;
private transient int rowIdFieldIndex;
+ private transient boolean multiColumn;
+ private transient ProjectedRow writerProjection;
BuildIndexOperator(
ReadBuilder readBuilder,
FileStoreTable table,
String indexType,
DataField indexField,
+ List<DataField> extraFields,
RowType projectedRowType,
Options mergedOptions) {
this.readBuilder = readBuilder;
this.table = table;
this.indexType = indexType;
this.indexField = indexField;
+ this.extraFields = extraFields;
this.projectedRowType = projectedRowType;
this.mergedOptions = mergedOptions;
}
@@ -575,10 +617,27 @@ public class GenericIndexTopoBuilder {
public void open() throws Exception {
super.open();
this.tableRead = readBuilder.newRead();
- this.indexFieldGetter =
- InternalRow.createFieldGetter(
- indexField.type(),
projectedRowType.getFieldIndex(indexField.name()));
+ // The primary column followed by the extra columns, in index
order. Field getters and
+ // the writer projection both need the full ordered list.
+ this.indexedFields = new ArrayList<>(1 + extraFields.size());
+ indexedFields.add(indexField);
+ indexedFields.addAll(extraFields);
+ this.indexFieldGetters = new
InternalRow.FieldGetter[indexedFields.size()];
+ for (int i = 0; i < indexedFields.size(); i++) {
+ DataField field = indexedFields.get(i);
+ indexFieldGetters[i] =
+ InternalRow.createFieldGetter(
+ field.type(),
projectedRowType.getFieldIndex(field.name()));
+ }
this.rowIdFieldIndex =
projectedRowType.getFieldIndex(SpecialFields.ROW_ID.name());
+ this.multiColumn = !extraFields.isEmpty();
+ if (multiColumn) {
+ int[] projection = new int[indexedFields.size()];
+ for (int i = 0; i < indexedFields.size(); i++) {
+ projection[i] =
projectedRowType.getFieldIndex(indexedFields.get(i).name());
+ }
+ this.writerProjection = ProjectedRow.from(projection);
+ }
}
@Override
@@ -595,9 +654,8 @@ public class GenericIndexTopoBuilder {
task.split.dataFiles().size());
long startTime = System.currentTimeMillis();
- GlobalIndexSingletonWriter indexWriter =
- (GlobalIndexSingletonWriter)
- createIndexWriter(table, indexType, indexField,
mergedOptions);
+ GlobalIndexWriter indexWriter =
+ createIndexWriter(table, indexType, indexField,
extraFields, mergedOptions);
try {
long rowsSeen = 0;
@@ -626,8 +684,14 @@ public class GenericIndexTopoBuilder {
}
// Only write rows within this shard's range
if (currentRowId >= task.shardRange.from) {
- Object fieldData =
indexFieldGetter.getFieldOrNull(row);
- indexWriter.write(fieldData);
+ if (multiColumn) {
+ long rowId = currentRowId -
task.shardRange.from;
+ ((GlobalIndexMultiColumnWriter) indexWriter)
+ .write(rowId,
writerProjection.replaceRow(row));
+ } else {
+ Object fieldData =
indexFieldGetters[0].getFieldOrNull(row);
+ ((GlobalIndexSingletonWriter)
indexWriter).write(fieldData);
+ }
rowsSeen++;
}
}
@@ -664,7 +728,7 @@ public class GenericIndexTopoBuilder {
table,
partition,
task.shardRange,
- indexField,
+ indexedFields,
indexType,
resultEntries);
output.collect(
@@ -688,7 +752,7 @@ public class GenericIndexTopoBuilder {
FileStoreTable table,
BinaryRow partition,
Range rowRange,
- DataField indexField,
+ List<DataField> indexFields,
String indexType,
List<ResultEntry> resultEntries)
throws IOException {
@@ -698,14 +762,14 @@ public class GenericIndexTopoBuilder {
table.store().pathFactory().globalIndexFileFactory(),
table.coreOptions(),
rowRange,
- indexField.id(),
+ indexFields,
indexType,
resultEntries);
return new CommitMessageImpl(
partition, 0, null, indexIncrement(indexFileMetas),
emptyIncrement());
}
- private static void closeWriterQuietly(GlobalIndexSingletonWriter writer) {
+ private static void closeWriterQuietly(GlobalIndexWriter writer) {
if (writer instanceof Closeable) {
try {
((Closeable) writer).close();
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateGlobalIndexProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateGlobalIndexProcedure.java
index ad62ad8f76..1979547777 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateGlobalIndexProcedure.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateGlobalIndexProcedure.java
@@ -20,10 +20,12 @@ package org.apache.paimon.flink.procedure;
import org.apache.paimon.flink.btree.BTreeIndexTopoBuilder;
import org.apache.paimon.flink.globalindex.GenericIndexTopoBuilder;
+import org.apache.paimon.globalindex.GlobalIndexer;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.ParameterUtils;
@@ -32,8 +34,11 @@ import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.ProcedureHint;
import org.apache.flink.table.procedure.ProcedureContext;
+import java.util.Arrays;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.stream.Collectors;
import static org.apache.paimon.utils.ParameterUtils.getPartitions;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -85,11 +90,23 @@ public class CreateGlobalIndexProcedure extends
ProcedureBase {
tableId);
RowType rowType = table.rowType();
+ List<String> indexColumns =
+ Arrays.stream(indexColumn.split(","))
+ .map(String::trim)
+ .filter(s -> !s.isEmpty())
+ .collect(Collectors.toList());
+ checkArgument(!indexColumns.isEmpty(), "At least one column
required.");
checkArgument(
- rowType.containsField(indexColumn),
- "Column '%s' does not exist in table '%s'.",
- indexColumn,
- tableId);
+ indexColumns.size() == new HashSet<>(indexColumns).size(),
+ "Duplicate index columns are not allowed: %s",
+ indexColumns);
+ for (String col : indexColumns) {
+ checkArgument(
+ rowType.containsField(col),
+ "Column '%s' does not exist in table '%s'.",
+ col,
+ tableId);
+ }
// Parse partition predicate
PartitionPredicate partitionPredicate = parsePartitionPredicate(table,
partitions);
@@ -99,12 +116,29 @@ public class CreateGlobalIndexProcedure extends
ProcedureBase {
// Build global index based on index type
indexType = indexType.toLowerCase().trim();
+ if (indexColumns.size() > 1) {
+ // Fail fast before submitting the job: index types that do not
support multi-column
+ // throw from GlobalIndexerFactory#create, which happens before
any indexer side effect.
+ DataField indexField = rowType.getField(indexColumns.get(0));
+ List<DataField> extraFields =
+ indexColumns.subList(1, indexColumns.size()).stream()
+ .map(rowType::getField)
+ .collect(Collectors.toList());
+ try {
+ GlobalIndexer.create(indexType, indexField, extraFields,
userOptions);
+ } catch (UnsupportedOperationException e) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Index type '%s' does not support multi-column
index, got columns: %s",
+ indexType, indexColumns));
+ }
+ }
try {
if ("btree".equals(indexType)) {
BTreeIndexTopoBuilder.buildIndexAndExecute(
procedureContext.getExecutionEnvironment(),
table,
- indexColumn,
+ indexColumns.get(0),
partitionPredicate,
userOptions);
return new String[] {
@@ -114,7 +148,8 @@ public class CreateGlobalIndexProcedure extends
ProcedureBase {
GenericIndexTopoBuilder.buildIndexAndExecute(
procedureContext.getExecutionEnvironment(),
table,
- indexColumn,
+ indexColumns.get(0),
+ indexColumns.subList(1, indexColumns.size()),
indexType,
partitionPredicate,
userOptions);
@@ -122,8 +157,8 @@ public class CreateGlobalIndexProcedure extends
ProcedureBase {
} catch (Exception e) {
throw new RuntimeException(
String.format(
- "Failed to create %s index for column '%s' on
table '%s'.",
- indexType, indexColumn, table.name()),
+ "Failed to create %s index for columns '%s' on
table '%s'.",
+ indexType, indexColumns, table.name()),
e);
}
return new String[] {
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/DropGlobalIndexProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/DropGlobalIndexProcedure.java
index a5ab0239c2..92bde693ea 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/DropGlobalIndexProcedure.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/DropGlobalIndexProcedure.java
@@ -42,6 +42,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@@ -82,13 +83,26 @@ public class DropGlobalIndexProcedure extends ProcedureBase
{
FileStoreTable table = (FileStoreTable) table(tableId);
- // Validate column exists
+ // Parse comma-separated columns (consistent with create procedure)
RowType rowType = table.rowType();
- checkArgument(
- rowType.containsField(indexColumn),
- "Column '%s' does not exist in table '%s'.",
- indexColumn,
- tableId);
+ List<String> indexColumns =
+ Arrays.stream(indexColumn.split(","))
+ .map(String::trim)
+ .filter(s -> !s.isEmpty())
+ .collect(Collectors.toList());
+ checkArgument(!indexColumns.isEmpty(), "At least one column
required.");
+ for (String col : indexColumns) {
+ checkArgument(
+ rowType.containsField(col),
+ "Column '%s' does not exist in table '%s'.",
+ col,
+ tableId);
+ }
+ final List<Integer> indexFieldIds =
+ indexColumns.stream()
+ .map(col -> rowType.getField(col).id())
+ .collect(Collectors.toList());
+ final String columnsDesc = String.join(",", indexColumns);
// Parse partition predicate
PartitionPredicate partitionPredicate = parsePartitionPredicate(table,
partitions);
@@ -96,9 +110,6 @@ public class DropGlobalIndexProcedure extends ProcedureBase {
// Normalize index type
final String indexTypeLower = indexType.toLowerCase().trim();
- // Get column field ID for final reference in lambda
- final int columnId = rowType.getField(indexColumn).id();
-
// Get latest snapshot
Snapshot snapshot =
table.latestSnapshot()
@@ -108,12 +119,15 @@ public class DropGlobalIndexProcedure extends
ProcedureBase {
String.format(
"Table '%s' has no
snapshot.", tableId)));
- // Create filter for index entries to delete
+ // Create filter for index entries to delete — match by primary column
+ full column set
Filter<IndexManifestEntry> filter =
entry ->
entry.indexFile().indexType().equals(indexTypeLower)
&& entry.indexFile().globalIndexMeta() != null
- &&
entry.indexFile().globalIndexMeta().indexFieldId() == columnId
+ && entry.indexFile()
+ .globalIndexMeta()
+ .getIndexedFieldIds()
+ .equals(indexFieldIds)
&& (partitionPredicate == null
||
partitionPredicate.test(entry.partition()));
@@ -122,15 +136,15 @@ public class DropGlobalIndexProcedure extends
ProcedureBase {
table.store().newIndexFileHandler().scan(snapshot, filter);
LOG.info(
- "Found {} {} global index files to delete for column '{}' on
table '{}'",
+ "Found {} {} global index files to delete for columns '{}' on
table '{}'",
waitToDelete.size(),
indexTypeLower,
- indexColumn,
+ columnsDesc,
table.name());
if (waitToDelete.isEmpty()) {
return new String[] {
- "No " + indexTypeLower + " global index found for column '" +
indexColumn + "'"
+ "No " + indexTypeLower + " global index found for columns '" +
columnsDesc + "'"
};
}
@@ -165,10 +179,10 @@ public class DropGlobalIndexProcedure extends
ProcedureBase {
}
LOG.info(
- "Successfully dropped {} {} global index files for column '{}'
on table '{}'",
+ "Successfully dropped {} {} global index files for columns
'{}' on table '{}'",
waitToDelete.size(),
indexTypeLower,
- indexColumn,
+ columnsDesc,
table.name());
return new String[] {
@@ -176,8 +190,8 @@ public class DropGlobalIndexProcedure extends ProcedureBase
{
+ waitToDelete.size()
+ " "
+ indexTypeLower
- + " global index files for column '"
- + indexColumn
+ + " global index files for columns '"
+ + columnsDesc
+ "' on table '"
+ table.name()
+ "'"
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilderTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilderTest.java
index 0de57077b2..c69b59ad6e 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilderTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilderTest.java
@@ -23,6 +23,7 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryRowWriter;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.globalindex.GlobalIndexBuilderUtils;
import org.apache.paimon.io.PojoDataFileMeta;
import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.ManifestEntry;
@@ -472,10 +473,10 @@ class GenericIndexTopoBuilderTest {
entries.add(createEntryWithSchemaId(BinaryRow.EMPTY_ROW, 200L, 100,
0L));
List<ManifestEntry> result =
- GenericIndexTopoBuilder.filterEntriesBefore(
+ GlobalIndexBuilderUtils.filterEntriesBefore(
entries,
- GenericIndexTopoBuilder.findMinNonIndexableRowId(
- schemaManager, entries, "vec"));
+ GlobalIndexBuilderUtils.findMinNonIndexableRowId(
+ schemaManager, entries,
Collections.singletonList("vec")));
assertThat(result).hasSize(2);
assertThat(result.get(0).file().nonNullFirstRowId()).isEqualTo(0L);
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/DropGlobalIndexProcedureITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/DropGlobalIndexProcedureITCase.java
index 5659467d8a..a348b5af7e 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/DropGlobalIndexProcedureITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/DropGlobalIndexProcedureITCase.java
@@ -299,6 +299,6 @@ public class DropGlobalIndexProcedureITCase extends
CatalogITCaseBase {
assertThat(dropResult.get(0).getField(0))
.isInstanceOf(String.class)
.asString()
- .contains("No btree global index found for column 'name'");
+ .contains("No btree global index found for columns 'name'");
}
}
diff --git
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
index ad6f5b9501..e5d7df44be 100644
---
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
+++
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
@@ -21,6 +21,7 @@ package org.apache.paimon.spark.commands
import org.apache.paimon.CoreOptions.GlobalIndexColumnUpdateAction
import org.apache.paimon.data.BinaryRow
import org.apache.paimon.format.blob.BlobFileFormat.isBlobFile
+import org.apache.paimon.index.GlobalIndexMeta
import org.apache.paimon.io.{CompactIncrement, DataIncrement}
import org.apache.paimon.manifest.IndexManifestEntry
import org.apache.paimon.spark.SparkTable
@@ -592,9 +593,9 @@ case class MergeIntoPaimonDataEvolutionTable(
if (globalIndexMeta == null) {
false
} else {
- val fieldName =
rowType.getField(globalIndexMeta.indexFieldId()).name()
+ val indexedNames =
globalIndexMeta.getIndexedFieldNames(rowType).asScala
affectedParts.contains(entry.partition()) && updateColumns.exists(
- _.name.equals(fieldName))
+ col => indexedNames.contains(col.name))
}
}
@@ -611,8 +612,7 @@ case class MergeIntoPaimonDataEvolutionTable(
case GlobalIndexColumnUpdateAction.THROW_ERROR =>
val updatedColNames = updateColumns.map(_.name)
val conflicted = affectedIndexEntries
- .map(_.indexFile().globalIndexMeta().indexFieldId())
- .map(id => rowType.getField(id).name())
+ .flatMap(e =>
e.indexFile().globalIndexMeta().getIndexedFieldNames(rowType).asScala)
.toSet
throw new RuntimeException(
s"""MergeInto: update columns contain globally indexed columns,
not supported now.
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexBuilder.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexBuilder.java
index 1485d14fac..ae87dc96a4 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexBuilder.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexBuilder.java
@@ -20,23 +20,29 @@ package org.apache.paimon.spark.globalindex;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.globalindex.GlobalIndexMultiColumnWriter;
import org.apache.paimon.globalindex.GlobalIndexSingletonWriter;
+import org.apache.paimon.globalindex.GlobalIndexWriter;
import org.apache.paimon.globalindex.ResultEntry;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataIncrement;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.SpecialFields;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.CloseableIterator;
import org.apache.paimon.utils.LongCounter;
+import org.apache.paimon.utils.ProjectedRow;
import org.apache.paimon.utils.Range;
import java.io.IOException;
import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import static
org.apache.paimon.globalindex.GlobalIndexBuilderUtils.createIndexWriter;
@@ -51,6 +57,7 @@ public class DefaultGlobalIndexBuilder implements
Serializable {
private final BinaryRow partition;
private final RowType readType;
private final DataField indexField;
+ private final List<DataField> extraFields;
private final String indexType;
private final Range rowRange;
private final Options options;
@@ -63,15 +70,48 @@ public class DefaultGlobalIndexBuilder implements
Serializable {
String indexType,
Range rowRange,
Options options) {
+ this(
+ table,
+ partition,
+ readType,
+ indexField,
+ Collections.emptyList(),
+ indexType,
+ rowRange,
+ options);
+ }
+
+ public DefaultGlobalIndexBuilder(
+ FileStoreTable table,
+ BinaryRow partition,
+ RowType readType,
+ DataField indexField,
+ List<DataField> extraFields,
+ String indexType,
+ Range rowRange,
+ Options options) {
this.table = table;
this.partition = partition;
this.readType = readType;
this.indexField = indexField;
+ // Copy into a serializable ArrayList: callers may pass a List#subList
view (e.g.
+ // indexFields.subList(1, ...)), which is not Serializable, and this
builder is serialized
+ // and shipped to Spark executors. A null value means no extra columns.
+ this.extraFields =
+ extraFields == null ? Collections.emptyList() : new
ArrayList<>(extraFields);
this.indexType = indexType;
this.rowRange = rowRange;
this.options = options;
}
+ /** The primary index column followed by the extra columns, in index
order. */
+ private List<DataField> indexedFields() {
+ List<DataField> fields = new ArrayList<>(1 + extraFields.size());
+ fields.add(indexField);
+ fields.addAll(extraFields);
+ return fields;
+ }
+
public FileStoreTable table() {
return table;
}
@@ -89,7 +129,7 @@ public class DefaultGlobalIndexBuilder implements
Serializable {
table.store().pathFactory().globalIndexFileFactory(),
table.coreOptions(),
rowRange,
- indexField.id(),
+ indexedFields(),
indexType,
resultEntries);
DataIncrement dataIncrement =
DataIncrement.indexIncrement(indexFileMetas);
@@ -99,27 +139,50 @@ public class DefaultGlobalIndexBuilder implements
Serializable {
private List<ResultEntry> writePaimonRows(
CloseableIterator<InternalRow> rows, LongCounter rowCounter)
throws IOException {
- GlobalIndexSingletonWriter indexWriter =
- (GlobalIndexSingletonWriter)
- createIndexWriter(table, indexType, indexField,
options);
+ GlobalIndexWriter indexWriter =
+ createIndexWriter(table, indexType, indexField, extraFields,
options);
+ boolean multiColumn = !extraFields.isEmpty();
try {
- InternalRow.FieldGetter getter =
- InternalRow.createFieldGetter(
- indexField.type(),
readType.getFieldIndex(indexField.name()));
- rows.forEachRemaining(
- row -> {
- Object indexO = getter.getFieldOrNull(row);
- indexWriter.write(indexO);
- rowCounter.add(1);
- });
+ if (multiColumn) {
+ GlobalIndexMultiColumnWriter multiWriter =
+ (GlobalIndexMultiColumnWriter) indexWriter;
+ List<DataField> indexedFields = indexedFields();
+ int[] projection = new int[indexedFields.size()];
+ for (int i = 0; i < indexedFields.size(); i++) {
+ DataField field = indexedFields.get(i);
+ projection[i] = readType.getFieldIndex(field.name());
+ }
+ ProjectedRow projectedRow = ProjectedRow.from(projection);
+ int rowIdIndex =
readType.getFieldIndex(SpecialFields.ROW_ID.name());
+ while (rows.hasNext()) {
+ InternalRow row = rows.next();
+ long absRowId = row.getLong(rowIdIndex);
+ if (absRowId < rowRange.from || absRowId > rowRange.to) {
+ continue;
+ }
+ multiWriter.write(absRowId - rowRange.from,
projectedRow.replaceRow(row));
+ rowCounter.add(1);
+ }
+ } else {
+ GlobalIndexSingletonWriter singleWriter =
(GlobalIndexSingletonWriter) indexWriter;
+ InternalRow.FieldGetter getter =
+ InternalRow.createFieldGetter(
+ indexField.type(),
readType.getFieldIndex(indexField.name()));
+ rows.forEachRemaining(
+ row -> {
+ Object indexO = getter.getFieldOrNull(row);
+ singleWriter.write(indexO);
+ rowCounter.add(1);
+ });
+ }
return indexWriter.finish();
} finally {
closeWriterQuietly(indexWriter);
}
}
- private static void closeWriterQuietly(GlobalIndexSingletonWriter writer) {
+ private static void closeWriterQuietly(GlobalIndexWriter writer) {
if (writer instanceof java.io.Closeable) {
try {
((java.io.Closeable) writer).close();
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilder.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilder.java
index d0c91c7479..4671006929 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilder.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilder.java
@@ -21,12 +21,14 @@ package org.apache.paimon.spark.globalindex;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.globalindex.GlobalIndexBuilderUtils;
import org.apache.paimon.globalindex.IndexedSplit;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageSerializer;
@@ -77,6 +79,30 @@ public class DefaultGlobalIndexTopoBuilder implements
GlobalIndexTopologyBuilder
DataField indexField,
Options options)
throws IOException {
+ return buildIndex(
+ spark,
+ relation,
+ partitionPredicate,
+ table,
+ indexType,
+ readType,
+ indexField,
+ Collections.emptyList(),
+ options);
+ }
+
+ @Override
+ public List<CommitMessage> buildIndex(
+ SparkSession spark,
+ DataSourceV2Relation relation,
+ PartitionPredicate partitionPredicate,
+ FileStoreTable table,
+ String indexType,
+ RowType readType,
+ DataField indexField,
+ List<DataField> extraFields,
+ Options options)
+ throws IOException {
Options tableOptions = table.coreOptions().toConfiguration();
long rowsPerShard =
tableOptions
@@ -88,6 +114,16 @@ public class DefaultGlobalIndexTopoBuilder implements
GlobalIndexTopologyBuilder
List<ManifestEntry> entries =
table.store().newScan().withPartitionFilter(partitionPredicate).plan().files();
+ List<DataField> indexFields = new ArrayList<>();
+ indexFields.add(indexField);
+ indexFields.addAll(extraFields);
+ List<String> indexColumns =
+
indexFields.stream().map(DataField::name).collect(Collectors.toList());
+ SchemaManager schemaManager = new SchemaManager(table.fileIO(),
table.location());
+ long boundaryRowId =
+ GlobalIndexBuilderUtils.findMinNonIndexableRowId(
+ schemaManager, entries, indexColumns);
+ entries = GlobalIndexBuilderUtils.filterEntriesBefore(entries,
boundaryRowId);
// generate splits for each partition && shard
Map<BinaryRow, List<IndexedSplit>> splits = split(table, entries,
rowsPerShard);
@@ -107,6 +143,7 @@ public class DefaultGlobalIndexTopoBuilder implements
GlobalIndexTopologyBuilder
partition,
readType,
indexField,
+ extraFields,
indexType,
indexedSplit.rowRanges().get(0),
options);
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/GlobalIndexTopologyBuilder.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/GlobalIndexTopologyBuilder.java
index 50c6ab34e1..d7a47cfdc9 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/GlobalIndexTopologyBuilder.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/GlobalIndexTopologyBuilder.java
@@ -46,4 +46,32 @@ public interface GlobalIndexTopologyBuilder {
DataField indexField,
Options options)
throws IOException;
+
+ default List<CommitMessage> buildIndex(
+ SparkSession spark,
+ DataSourceV2Relation relation,
+ PartitionPredicate partitionPredicate,
+ FileStoreTable table,
+ String indexType,
+ RowType readType,
+ DataField indexField,
+ List<DataField> extraFields,
+ Options options)
+ throws IOException {
+ if (extraFields != null && !extraFields.isEmpty()) {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Topology builder '%s' does not support
multi-column index, got extra columns: %s",
+ identifier(), extraFields));
+ }
+ return buildIndex(
+ spark,
+ relation,
+ partitionPredicate,
+ table,
+ indexType,
+ readType,
+ indexField,
+ options);
+ }
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedure.java
index b447cdbd33..adceb783af 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedure.java
@@ -18,6 +18,7 @@
package org.apache.paimon.spark.procedure;
+import org.apache.paimon.globalindex.GlobalIndexer;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.spark.globalindex.GlobalIndexTopologyBuilder;
@@ -43,12 +44,14 @@ import org.apache.spark.sql.types.StructType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.Collections;
+import java.util.Arrays;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.UUID;
+import java.util.stream.Collectors;
import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.spark.sql.types.DataTypes.StringType;
@@ -120,6 +123,11 @@ public class CreateGlobalIndexProcedure extends
BaseProcedure {
return modifySparkTable(
tableIdent,
sparkTable -> {
+ List<String> indexColumns =
+ Arrays.stream(column.split(","))
+ .map(String::trim)
+ .filter(s -> !s.isEmpty())
+ .collect(Collectors.toList());
try {
org.apache.paimon.table.Table t =
sparkTable.getTable();
checkArgument(
@@ -132,11 +140,24 @@ public class CreateGlobalIndexProcedure extends
BaseProcedure {
tableIdent);
RowType rowType = table.rowType();
+ checkArgument(!indexColumns.isEmpty(), "At least one
column required.");
checkArgument(
- rowType.containsField(column),
- "Column '%s' does not exist in table '%s'.",
- column,
- tableIdent);
+ indexColumns.size() == new
HashSet<>(indexColumns).size(),
+ "Duplicate index columns are not allowed: %s",
+ indexColumns);
+ // No hard cap on the number of index columns: unlike
row-store B-tree
+ // indexes (e.g. MySQL 16, PostgreSQL 32) whose limit
comes from composing
+ // columns into a single key, the global index is
built on per-type index
+ // frameworks. Whether multiple columns are supported,
and any practical
+ // limit, is decided by each index type (single-column
types reject
+ // multi-column via UnsupportedOperationException).
+ for (String col : indexColumns) {
+ checkArgument(
+ rowType.containsField(col),
+ "Column '%s' does not exist in table
'%s'.",
+ col,
+ tableIdent);
+ }
DataSourceV2Relation relation =
createRelation(tableIdent, sparkTable);
PartitionPredicate partitionPredicate =
SparkProcedureUtils.convertToPartitionPredicate(
@@ -145,13 +166,33 @@ public class CreateGlobalIndexProcedure extends
BaseProcedure {
spark(),
relation);
- DataField indexField = rowType.getField(column);
- RowType projectedRowType =
-
rowType.project(Collections.singletonList(column));
+ List<DataField> indexFields =
+ indexColumns.stream()
+ .map(rowType::getField)
+ .collect(Collectors.toList());
+ RowType projectedRowType =
rowType.project(indexColumns);
RowType readRowType =
SpecialFields.rowTypeWithRowId(projectedRowType);
Options userOptions = createUserOptions(table,
optionString);
+ if (indexColumns.size() > 1) {
+ // Fail fast before submitting the job: index
types that do not support
+ // multi-column throw from
GlobalIndexerFactory#create, which happens
+ // before any indexer side effect.
+ try {
+ GlobalIndexer.create(
+ indexType,
+ indexFields.get(0),
+ indexFields.subList(1,
indexFields.size()),
+ userOptions);
+ } catch (UnsupportedOperationException e) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Index type '%s' does not
support multi-column index, got columns: %s",
+ indexType, indexColumns));
+ }
+ }
+
GlobalIndexTopologyBuilder topoBuilder =
GlobalIndexTopologyBuilderUtils.createTopoBuilder(indexType);
@@ -163,7 +204,8 @@ public class CreateGlobalIndexProcedure extends
BaseProcedure {
table,
indexType,
readRowType,
- indexField,
+ indexFields.get(0),
+ indexFields.subList(1,
indexFields.size()),
userOptions);
try (TableCommitImpl commit =
@@ -179,8 +221,8 @@ public class CreateGlobalIndexProcedure extends
BaseProcedure {
} catch (Exception e) {
throw new RuntimeException(
String.format(
- "Failed to create %s index for column
'%s' on table '%s'.",
- indexType, column, tableIdent),
+ "Failed to create %s index for columns
'%s' on table '%s'.",
+ indexType, indexColumns, tableIdent),
e);
}
});
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DropGlobalIndexProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DropGlobalIndexProcedure.java
index 74e4cc4aea..bd218eb68d 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DropGlobalIndexProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DropGlobalIndexProcedure.java
@@ -46,6 +46,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.Locale;
import java.util.Map;
@@ -107,6 +108,13 @@ public class DropGlobalIndexProcedure extends
BaseProcedure {
LOG.info("Starting to drop index for table " + tableIdent + " WHERE: "
+ finalWhere);
+ List<String> indexColumns =
+ Arrays.stream(column.split(","))
+ .map(String::trim)
+ .filter(s -> !s.isEmpty())
+ .collect(Collectors.toList());
+ checkArgument(!indexColumns.isEmpty(), "At least one column
required.");
+
return modifyPaimonTable(
tableIdent,
t -> {
@@ -117,11 +125,17 @@ public class DropGlobalIndexProcedure extends
BaseProcedure {
FileStoreTable table = (FileStoreTable) t;
RowType rowType = table.rowType();
- checkArgument(
- rowType.containsField(column),
- "Column '%s' does not exist in table '%s'.",
- column,
- tableIdent);
+ for (String col : indexColumns) {
+ checkArgument(
+ rowType.containsField(col),
+ "Column '%s' does not exist in table
'%s'.",
+ col,
+ tableIdent);
+ }
+ List<Integer> indexFieldIds =
+ indexColumns.stream()
+ .map(col -> rowType.getField(col).id())
+ .collect(Collectors.toList());
DataSourceV2Relation relation =
createRelation(tableIdent);
PartitionPredicate partitionPredicate =
SparkProcedureUtils.convertToPartitionPredicate(
@@ -144,9 +158,9 @@ public class DropGlobalIndexProcedure extends BaseProcedure
{
entry.indexFile().indexType().equals(indexType)
&&
entry.indexFile().globalIndexMeta() != null
&& entry.indexFile()
-
.globalIndexMeta()
- .indexFieldId()
- ==
rowType.getField(column).id()
+ .globalIndexMeta()
+ .getIndexedFieldIds()
+ .equals(indexFieldIds)
&& (partitionPredicate == null
||
partitionPredicate.test(
entry.partition()));
@@ -192,8 +206,8 @@ public class DropGlobalIndexProcedure extends BaseProcedure
{
} catch (Exception e) {
throw new RuntimeException(
String.format(
- "Failed to drop %s index for column
'%s' on table '%s'.",
- indexType, column, tableIdent),
+ "Failed to drop %s index for columns
'%s' on table '%s'.",
+ indexType, String.join(",",
indexColumns), tableIdent),
e);
}
});
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
index cd1b000a36..99990637c3 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
@@ -21,6 +21,7 @@ package org.apache.paimon.spark.commands
import org.apache.paimon.CoreOptions.GlobalIndexColumnUpdateAction
import org.apache.paimon.data.BinaryRow
import org.apache.paimon.format.blob.BlobFileFormat.isBlobFile
+import org.apache.paimon.index.GlobalIndexMeta
import org.apache.paimon.io.{CompactIncrement, DataIncrement}
import org.apache.paimon.manifest.IndexManifestEntry
import org.apache.paimon.spark.SparkTable
@@ -594,9 +595,9 @@ case class MergeIntoPaimonDataEvolutionTable(
if (globalIndexMeta == null) {
false
} else {
- val fieldName =
rowType.getField(globalIndexMeta.indexFieldId()).name()
+ val indexedNames =
globalIndexMeta.getIndexedFieldNames(rowType).asScala
affectedParts.contains(entry.partition()) && updateColumns.exists(
- _.name.equals(fieldName))
+ col => indexedNames.contains(col.name))
}
}
@@ -613,8 +614,7 @@ case class MergeIntoPaimonDataEvolutionTable(
case GlobalIndexColumnUpdateAction.THROW_ERROR =>
val updatedColNames = updateColumns.map(_.name)
val conflicted = affectedIndexEntries
- .map(_.indexFile().globalIndexMeta().indexFieldId())
- .map(id => rowType.getField(id).name())
+ .flatMap(e =>
e.indexFile().globalIndexMeta().getIndexedFieldNames(rowType).asScala)
.toSet
throw new RuntimeException(
s"""MergeInto: update columns contain globally indexed columns,
not supported now.
