From 56f4d21cb29cb6b12a8e6b252e13f62836251928 Mon Sep 17 00:00:00 2001 From: zhangjunfan Date: Wed, 15 Jul 2026 17:53:30 +0800 Subject: [PATCH 1/2] [core] add data file table query support w/o local cache --- .../privilege/PrivilegedFileStoreTable.java | 7 + .../table/AppendOnlyFileStoreTable.java | 6 + .../paimon/table/DelegatedFileStoreTable.java | 6 + .../apache/paimon/table/FileStoreTable.java | 3 + .../table/PrimaryKeyFileStoreTable.java | 6 + .../table/query/DataFileTableQuery.java | 255 ++++++++++++++++++ .../table/PrimaryKeySimpleTableTest.java | 67 +++++ 7 files changed, 350 insertions(+) create mode 100644 paimon-core/src/main/java/org/apache/paimon/table/query/DataFileTableQuery.java diff --git a/paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStoreTable.java index e1064e70458a..486a20eaa8a4 100644 --- a/paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStoreTable.java @@ -26,6 +26,7 @@ import org.apache.paimon.table.DelegatedFileStoreTable; import org.apache.paimon.table.ExpireSnapshots; import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.query.DataFileTableQuery; import org.apache.paimon.table.query.LocalTableQuery; import org.apache.paimon.table.sink.TableCommitImpl; import org.apache.paimon.table.sink.TableWriteImpl; @@ -276,6 +277,12 @@ public LocalTableQuery newLocalTableQuery() { return wrapped.newLocalTableQuery(); } + @Override + public DataFileTableQuery newDataFileTableQuery() { + privilegeChecker.assertCanSelect(identifier); + return wrapped.newDataFileTableQuery(); + } + // ======================= equals ============================ @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java index 4d35147c17a4..5accbfd63c99 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java @@ -29,6 +29,7 @@ import org.apache.paimon.operation.FileStoreScan; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.query.DataFileTableQuery; import org.apache.paimon.table.query.LocalTableQuery; import org.apache.paimon.table.sink.TableWriteImpl; import org.apache.paimon.table.source.AppendBatchTableScan; @@ -181,4 +182,9 @@ public TableWriteImpl newWrite(String commitUser, @Nullable Integer public LocalTableQuery newLocalTableQuery() { throw new UnsupportedOperationException(); } + + @Override + public DataFileTableQuery newDataFileTableQuery() { + throw new UnsupportedOperationException(); + } } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/DelegatedFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/DelegatedFileStoreTable.java index f9bcbeee01c0..9ef2d839c226 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/DelegatedFileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/DelegatedFileStoreTable.java @@ -30,6 +30,7 @@ import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.schema.TableSchema; import org.apache.paimon.stats.Statistics; +import org.apache.paimon.table.query.DataFileTableQuery; import org.apache.paimon.table.query.LocalTableQuery; import org.apache.paimon.table.sink.RowKeyExtractor; import org.apache.paimon.table.sink.TableCommitImpl; @@ -355,6 +356,11 @@ public LocalTableQuery newLocalTableQuery() { return wrapped.newLocalTableQuery(); } + @Override + public DataFileTableQuery newDataFileTableQuery() { + return wrapped.newDataFileTableQuery(); + } + @Override public boolean supportStreamingReadOverwrite() { return wrapped.supportStreamingReadOverwrite(); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/FileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/FileStoreTable.java index 5a3f87d0edf1..b8bdd55dd939 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/FileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/FileStoreTable.java @@ -27,6 +27,7 @@ import org.apache.paimon.options.ExpireConfig; import org.apache.paimon.schema.TableSchema; import org.apache.paimon.stats.Statistics; +import org.apache.paimon.table.query.DataFileTableQuery; import org.apache.paimon.table.query.LocalTableQuery; import org.apache.paimon.table.sink.BatchTableCommit; import org.apache.paimon.table.sink.RowKeyExtractor; @@ -138,6 +139,8 @@ default Optional comment() { LocalTableQuery newLocalTableQuery(); + DataFileTableQuery newDataFileTableQuery(); + boolean supportStreamingReadOverwrite(); RowKeyExtractor createRowKeyExtractor(); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java index 3030b3504e1d..86ec3862c031 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java @@ -31,6 +31,7 @@ import org.apache.paimon.predicate.Predicate; import org.apache.paimon.schema.KeyValueFieldsExtractor; import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.query.DataFileTableQuery; import org.apache.paimon.table.query.LocalTableQuery; import org.apache.paimon.table.sink.TableWriteImpl; import org.apache.paimon.table.source.DataTableScan; @@ -193,6 +194,11 @@ public LocalTableQuery newLocalTableQuery() { return new LocalTableQuery(this); } + @Override + public DataFileTableQuery newDataFileTableQuery() { + return new DataFileTableQuery(this); + } + @Override @Nullable protected Runnable newExpireRunnable() { diff --git a/paimon-core/src/main/java/org/apache/paimon/table/query/DataFileTableQuery.java b/paimon-core/src/main/java/org/apache/paimon/table/query/DataFileTableQuery.java new file mode 100644 index 000000000000..9879354abf56 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/table/query/DataFileTableQuery.java @@ -0,0 +1,255 @@ +/* + * 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.query; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.FileStore; +import org.apache.paimon.KeyValue; +import org.apache.paimon.KeyValueFileStore; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.data.serializer.InternalRowSerializer; +import org.apache.paimon.data.serializer.InternalSerializers; +import org.apache.paimon.deletionvectors.DeletionVector; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.io.KeyValueFileReaderFactory; +import org.apache.paimon.mergetree.Levels; +import org.apache.paimon.mergetree.LookupUtils; +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.types.DataField; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.Filter; +import org.apache.paimon.utils.KeyComparatorSupplier; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; + +/** + * Implementation for {@link TableQuery} which looks up records directly from data files. + * + *

This query pushes primary key predicates down to the file format reader and does not build or + * cache local lookup files. Callers are responsible for keeping the data file view up to date with + * {@link #refreshFiles(BinaryRow, int, List, List)}. + */ +public class DataFileTableQuery implements TableQuery { + + private final Map> tableView; + private final CoreOptions options; + private final Comparator keyComparator; + private final KeyValueFileReaderFactory.Builder readerFactoryBuilder; + private final RowType rowType; + private final List primaryKeyFieldIndexes; + private final InternalRow.FieldGetter[] keyFieldGetters; + private final int startLevel; + + @Nullable private Filter rowFilter; + + public DataFileTableQuery(FileStoreTable table) { + this.options = table.coreOptions(); + this.tableView = new ConcurrentHashMap<>(); + + FileStore tableStore = table.store(); + if (!(tableStore instanceof KeyValueFileStore)) { + throw new UnsupportedOperationException( + "Table Query only supports table with primary key."); + } + KeyValueFileStore store = (KeyValueFileStore) tableStore; + + this.readerFactoryBuilder = store.newReaderFactoryBuilder(); + this.keyComparator = new KeyComparatorSupplier(readerFactoryBuilder.keyType()).get(); + this.rowType = table.schema().logicalRowType(); + List primaryKeyFields = table.schema().trimmedPrimaryKeysFields(); + this.primaryKeyFieldIndexes = new ArrayList<>(primaryKeyFields.size()); + this.keyFieldGetters = new InternalRow.FieldGetter[primaryKeyFields.size()]; + for (int i = 0; i < primaryKeyFields.size(); i++) { + DataField field = primaryKeyFields.get(i); + primaryKeyFieldIndexes.add(rowType.getFieldNames().indexOf(field.name())); + keyFieldGetters[i] = InternalRow.createFieldGetter(field.type(), i); + } + this.startLevel = options.needLookup() ? 1 : 0; + } + + /** Refreshes the data files of a partition and bucket. */ + public void refreshFiles( + BinaryRow partition, + int bucket, + List beforeFiles, + List dataFiles) { + BucketQueryState state = + tableView + .computeIfAbsent(partition, k -> new ConcurrentHashMap<>()) + .computeIfAbsent(bucket, k -> new BucketQueryState()); + state.lock.writeLock().lock(); + try { + if (state.levels == null) { + // Initial phase: ignore beforeFiles as they represent deletions from previous state + state.levels = new Levels(keyComparator, dataFiles, options.numLevels()); + } else { + state.levels.update(beforeFiles, dataFiles); + } + } finally { + state.lock.writeLock().unlock(); + } + } + + @Nullable + @Override + public InternalRow lookup(BinaryRow partition, int bucket, InternalRow key) throws IOException { + Map buckets = tableView.get(partition); + if (buckets == null || buckets.isEmpty()) { + return null; + } + BucketQueryState state = buckets.get(bucket); + if (state == null) { + return null; + } + + state.lock.readLock().lock(); + try { + if (state.levels == null) { + return null; + } + + KeyValue kv = lookup(partition, bucket, state.levels, key); + if (kv == null + || kv.valueKind().isRetract() + || (rowFilter != null && !rowFilter.test(kv.value()))) { + return null; + } + return kv.value(); + } finally { + state.lock.readLock().unlock(); + } + } + + @Nullable + private KeyValue lookup(BinaryRow partition, int bucket, Levels levels, InternalRow key) + throws IOException { + return LookupUtils.lookup( + levels, + key, + startLevel, + (target, level) -> + LookupUtils.lookup( + keyComparator, + target, + level, + (lookupKey, file) -> lookup(partition, bucket, lookupKey, file)), + (target, level0) -> + LookupUtils.lookupLevel0( + keyComparator, + target, + level0, + (lookupKey, file) -> lookup(partition, bucket, lookupKey, file))); + } + + @Nullable + private KeyValue lookup(BinaryRow partition, int bucket, InternalRow key, DataFileMeta file) + throws IOException { + KeyValueFileReaderFactory readerFactory = + readerFactoryBuilder.build( + partition, + bucket, + DeletionVector.emptyFactory(), + true, + primaryKeyPredicates(key)); + InternalRowSerializer valueSerializer = createValueSerializer(); + try (RecordReader reader = readerFactory.createRecordReader(file)) { + RecordReader.RecordIterator batch; + while ((batch = reader.readBatch()) != null) { + KeyValue result = null; + try { + KeyValue kv; + while ((kv = batch.next()) != null) { + if (keyComparator.compare(key, kv.key()) == 0) { + result = + new KeyValue() + .replace( + key, + kv.sequenceNumber(), + kv.valueKind(), + valueSerializer.copy(kv.value())) + .setLevel(kv.level()); + break; + } + } + } finally { + batch.releaseBatch(); + } + if (result != null) { + return result; + } + } + } + return null; + } + + private List primaryKeyPredicates(InternalRow key) { + PredicateBuilder builder = new PredicateBuilder(rowType); + List predicates = new ArrayList<>(primaryKeyFieldIndexes.size()); + for (int i = 0; i < primaryKeyFieldIndexes.size(); i++) { + int fieldIndex = primaryKeyFieldIndexes.get(i); + Object literal = keyFieldGetters[i].getFieldOrNull(key); + predicates.add( + literal == null + ? builder.isNull(fieldIndex) + : builder.equal(fieldIndex, literal)); + } + return predicates; + } + + @Override + public DataFileTableQuery withValueProjection(int[] projection) { + this.readerFactoryBuilder.withReadValueType(rowType.project(projection)); + return this; + } + + public DataFileTableQuery withRowFilter(Filter rowFilter) { + this.rowFilter = rowFilter; + return this; + } + + @Override + public InternalRowSerializer createValueSerializer() { + return InternalSerializers.create(readerFactoryBuilder.readValueType()); + } + + @Override + public void close() { + tableView.clear(); + } + + private static class BucketQueryState { + + private final ReadWriteLock lock = new ReentrantReadWriteLock(); + + @Nullable private Levels levels; + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java index dd3a21b86c2c..6b314c45d8d0 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java @@ -51,6 +51,7 @@ import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.schema.SchemaUtils; import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.query.DataFileTableQuery; import org.apache.paimon.table.query.LocalTableQuery; import org.apache.paimon.table.sink.BatchTableCommit; import org.apache.paimon.table.sink.BatchTableWrite; @@ -2324,6 +2325,55 @@ public void testTableQueryForNormal() throws Exception { innerTestTableQuery(table); } + @Test + public void testDataFileTableQuery() throws Exception { + FileStoreTable table = + createFileStoreTable(options -> options.set(FILE_FORMAT, FILE_FORMAT_PARQUET)); + StreamTableWrite write = table.newWrite(commitUser); + StreamTableCommit commit = table.newCommit(commitUser); + + write.write(rowData(1, 10, 100L)); + write.write(rowData(1, 20, 200L)); + List commitMessages = write.prepareCommit(true, 0); + commit.commit(0, commitMessages); + + DataFileTableQuery query = + table.newDataFileTableQuery().withValueProjection(new int[] {2, 1, 0}); + refreshDataFileTableQuery(query, commitMessages); + + InternalRow value = query.lookup(row(1), 0, row(10)); + assertThat(value).isNotNull(); + assertThat(value.getLong(0)).isEqualTo(100L); + assertThat(value.getInt(1)).isEqualTo(10); + assertThat(value.getInt(2)).isEqualTo(1); + assertThat(query.lookup(row(1), 0, row(30))).isNull(); + + query.withRowFilter(row -> row.getLong(0) >= 200L); + assertThat(query.lookup(row(1), 0, row(10))).isNull(); + assertThat(query.lookup(row(1), 0, row(20))).isNotNull(); + + write.write(rowData(1, 10, 300L)); + commitMessages = write.prepareCommit(true, 1); + commit.commit(1, commitMessages); + refreshDataFileTableQuery(query, commitMessages); + value = query.lookup(row(1), 0, row(10)); + assertThat(value).isNotNull(); + assertThat(value.getLong(0)).isEqualTo(300L); + query.withRowFilter(row -> row.getLong(0) < 200L); + assertThat(query.lookup(row(1), 0, row(10))).isNull(); + + query.withRowFilter(row -> true); + write.write(rowDataWithKind(RowKind.DELETE, 1, 10, 300L)); + commitMessages = write.prepareCommit(true, 2); + commit.commit(2, commitMessages); + refreshDataFileTableQuery(query, commitMessages); + assertThat(query.lookup(row(1), 0, row(10))).isNull(); + + query.close(); + write.close(); + commit.close(); + } + @Test public void testTableQueryDownloadsRemoteLookupFile() throws Exception { // Writer persists remote lookup ssts (deletion-vectors off -> "value" processor). @@ -2708,6 +2758,23 @@ private void refreshTableService(LocalTableQuery query, List comm } } + private void refreshDataFileTableQuery( + DataFileTableQuery query, List commitMessages) { + for (CommitMessage m : commitMessages) { + CommitMessageImpl msg = (CommitMessageImpl) m; + query.refreshFiles( + msg.partition(), + msg.bucket(), + msg.newFilesIncrement().deletedFiles(), + msg.newFilesIncrement().newFiles()); + query.refreshFiles( + msg.partition(), + msg.bucket(), + msg.compactIncrement().compactBefore(), + msg.compactIncrement().compactAfter()); + } + } + @Test public void writeMultiplePartitions() throws Exception { testWritePreemptMemory(false); From 0ce9af6c23e15535629f216b056ae1f9f27a5835 Mon Sep 17 00:00:00 2001 From: zhangjunfan Date: Thu, 16 Jul 2026 10:23:06 +0800 Subject: [PATCH 2/2] fix --- .../table/query/DataFileTableQuery.java | 63 ++++++++++++------- .../table/PrimaryKeySimpleTableTest.java | 45 +++++++++++-- 2 files changed, 80 insertions(+), 28 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/query/DataFileTableQuery.java b/paimon-core/src/main/java/org/apache/paimon/table/query/DataFileTableQuery.java index 9879354abf56..bf6b54f40331 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/query/DataFileTableQuery.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/query/DataFileTableQuery.java @@ -48,8 +48,8 @@ import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.locks.ReadWriteLock; -import java.util.concurrent.locks.ReentrantReadWriteLock; + +import static org.apache.paimon.CoreOptions.MergeEngine.DEDUPLICATE; /** * Implementation for {@link TableQuery} which looks up records directly from data files. @@ -57,6 +57,9 @@ *

This query pushes primary key predicates down to the file format reader and does not build or * cache local lookup files. Callers are responsible for keeping the data file view up to date with * {@link #refreshFiles(BinaryRow, int, List, List)}. + * + *

For tables which do not enable lookup, this query only supports the deduplicate merge engine + * without sequence fields. Deletion vectors are not supported. */ public class DataFileTableQuery implements TableQuery { @@ -75,6 +78,23 @@ public DataFileTableQuery(FileStoreTable table) { this.options = table.coreOptions(); this.tableView = new ConcurrentHashMap<>(); + if (options.deletionVectorsEnabled()) { + throw new UnsupportedOperationException( + "Data file table query does not support deletion vectors."); + } + if (!options.needLookup() && options.mergeEngine() != DEDUPLICATE) { + throw new UnsupportedOperationException( + "Data file table query only supports deduplicate merge engine when lookup is " + + "disabled, but merge engine is: " + + options.mergeEngine()); + } + if (options.mergeEngine() == DEDUPLICATE && !options.sequenceField().isEmpty()) { + throw new UnsupportedOperationException( + "Data file table query does not support sequence fields for deduplicate merge " + + "engine, but sequence fields are: " + + options.sequenceField()); + } + FileStore tableStore = table.store(); if (!(tableStore instanceof KeyValueFileStore)) { throw new UnsupportedOperationException( @@ -106,16 +126,17 @@ public void refreshFiles( tableView .computeIfAbsent(partition, k -> new ConcurrentHashMap<>()) .computeIfAbsent(bucket, k -> new BucketQueryState()); - state.lock.writeLock().lock(); - try { + synchronized (state) { if (state.levels == null) { // Initial phase: ignore beforeFiles as they represent deletions from previous state state.levels = new Levels(keyComparator, dataFiles, options.numLevels()); } else { - state.levels.update(beforeFiles, dataFiles); + // Publish a new immutable view so that lookups do not hold locks during file IO. + Levels levels = + new Levels(keyComparator, state.levels.allFiles(), options.numLevels()); + levels.update(beforeFiles, dataFiles); + state.levels = levels; } - } finally { - state.lock.writeLock().unlock(); } } @@ -131,22 +152,18 @@ public InternalRow lookup(BinaryRow partition, int bucket, InternalRow key) thro return null; } - state.lock.readLock().lock(); - try { - if (state.levels == null) { - return null; - } + Levels levels = state.levels; + if (levels == null) { + return null; + } - KeyValue kv = lookup(partition, bucket, state.levels, key); - if (kv == null - || kv.valueKind().isRetract() - || (rowFilter != null && !rowFilter.test(kv.value()))) { - return null; - } - return kv.value(); - } finally { - state.lock.readLock().unlock(); + KeyValue kv = lookup(partition, bucket, levels, key); + if (kv == null + || kv.valueKind().isRetract() + || (rowFilter != null && !rowFilter.test(kv.value()))) { + return null; } + return kv.value(); } @Nullable @@ -248,8 +265,6 @@ public void close() { private static class BucketQueryState { - private final ReadWriteLock lock = new ReentrantReadWriteLock(); - - @Nullable private Levels levels; + @Nullable private volatile Levels levels; } } diff --git a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java index 6b314c45d8d0..0ce50beb2ff8 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java @@ -2337,15 +2337,12 @@ public void testDataFileTableQuery() throws Exception { List commitMessages = write.prepareCommit(true, 0); commit.commit(0, commitMessages); - DataFileTableQuery query = - table.newDataFileTableQuery().withValueProjection(new int[] {2, 1, 0}); + DataFileTableQuery query = table.newDataFileTableQuery().withValueProjection(new int[] {2}); refreshDataFileTableQuery(query, commitMessages); InternalRow value = query.lookup(row(1), 0, row(10)); assertThat(value).isNotNull(); assertThat(value.getLong(0)).isEqualTo(100L); - assertThat(value.getInt(1)).isEqualTo(10); - assertThat(value.getInt(2)).isEqualTo(1); assertThat(query.lookup(row(1), 0, row(30))).isNull(); query.withRowFilter(row -> row.getLong(0) >= 200L); @@ -2374,6 +2371,32 @@ public void testDataFileTableQuery() throws Exception { commit.close(); } + @ParameterizedTest + @ValueSource(strings = {"partial-update", "sequence-field", "deletion-vectors"}) + public void testDataFileTableQueryUnsupportedOptions(String unsupportedOption) + throws Exception { + FileStoreTable table = + createFileStoreTable( + options -> { + switch (unsupportedOption) { + case "partial-update": + options.set(MERGE_ENGINE, PARTIAL_UPDATE); + break; + case "sequence-field": + options.set(CoreOptions.SEQUENCE_FIELD, "b"); + break; + case "deletion-vectors": + options.set(DELETION_VECTORS_ENABLED, true); + break; + default: + throw new IllegalArgumentException(unsupportedOption); + } + }); + + assertThatThrownBy(table::newDataFileTableQuery) + .isInstanceOf(UnsupportedOperationException.class); + } + @Test public void testTableQueryDownloadsRemoteLookupFile() throws Exception { // Writer persists remote lookup ssts (deletion-vectors off -> "value" processor). @@ -2631,6 +2654,20 @@ public void testForceLookupCompaction(CoreOptions.MergeEngine mergeEngine) throw .isEqualTo( Collections.singletonList( "1|10|100|binary|varbinary|mapKey:mapVal|multiset")); + + DataFileTableQuery query = table.newDataFileTableQuery().withValueProjection(new int[] {2}); + for (Split split : splits) { + DataSplit dataSplit = (DataSplit) split; + query.refreshFiles( + dataSplit.partition(), + dataSplit.bucket(), + Collections.emptyList(), + dataSplit.dataFiles()); + } + InternalRow value = query.lookup(row(1), 0, row(10)); + assertThat(value).isNotNull(); + assertThat(value.getLong(0)).isEqualTo(100L); + query.close(); } @Test