This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new fcae20748a [core] Add Multivalue global index primitives (#9288)
fcae20748a is described below
commit fcae20748a777fa72763d7a0f046482cc9ceb6f8
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Aug 18 17:14:39 2026 +0800
[core] Add Multivalue global index primitives (#9288)
---
docs/docs/program-api/java-api.mdx | 1 +
.../apache/paimon/fileindex/FileIndexReader.java | 5 +
.../globalindex/ConstantGlobalIndexReader.java | 6 +
.../globalindex/GlobalIndexKeyExtractor.java | 80 ++++++
.../paimon/globalindex/GlobalIndexReader.java | 6 +
.../globalindex/GlobalIndexSingleColumnWriter.java | 12 +
.../globalindex/OffsetGlobalIndexReader.java | 6 +
.../paimon/globalindex/SortedFileMetaSelector.java | 5 +
...eColumnWriter.java => SortedGlobalIndexer.java} | 15 +-
.../paimon/globalindex/UnionGlobalIndexReader.java | 6 +
.../globalindex/bitmap/BitmapGlobalIndexer.java | 11 +-
.../MultiValueBitmapIndexReader.java} | 100 +++++---
.../bitmap/MultiValueBitmapIndexWriter.java | 167 +++++++++++++
.../bitmap/MultiValueGlobalIndexOptions.java | 47 ++++
.../bitmap/MultiValueGlobalIndexer.java | 127 ++++++++++
.../MultiValueGlobalIndexerFactory.java} | 29 ++-
.../bitmap/MultiValueIndexFileMeta.java | 73 ++++++
.../globalindex/btree/BTreeGlobalIndexer.java | 11 +-
.../org/apache/paimon/predicate/ArrayContains.java | 90 +++++++
.../apache/paimon/predicate/FunctionVisitor.java | 4 +
.../org/apache/paimon/predicate/LeafFunction.java | 6 +
.../org/apache/paimon/predicate/LeafPredicate.java | 7 +-
.../predicate/OnlyPartitionKeyEqualVisitor.java | 5 +
.../apache/paimon/predicate/PredicateBuilder.java | 11 +
.../globalindex/GlobalIndexEvaluatorTest.java | 85 +++++++
.../bitmap/MultiValueBitmapIndexReaderTest.java | 273 +++++++++++++++++++++
.../apache/paimon/predicate/LeafPredicateTest.java | 22 ++
.../paimon/predicate/PredicateBuilderTest.java | 39 +++
.../paimon/predicate/PredicateJsonSerdeTest.java | 12 +-
.../orc/filter/OrcPredicateFunctionVisitor.java | 5 +
.../parquet/filter2/predicate/ParquetFilters.java | 5 +
31 files changed, 1209 insertions(+), 62 deletions(-)
diff --git a/docs/docs/program-api/java-api.mdx
b/docs/docs/program-api/java-api.mdx
index 887f36c1bb..0f58070037 100644
--- a/docs/docs/program-api/java-api.mdx
+++ b/docs/docs/program-api/java-api.mdx
@@ -477,3 +477,4 @@ public class StreamWriteTable {
| >= | org.apache.paimon.predicate.PredicateBuilder.greaterOrEqual |
| between | org.apache.paimon.predicate.PredicateBuilder.between |
| like | org.apache.paimon.predicate.PredicateBuilder.like |
+| array contains | org.apache.paimon.predicate.PredicateBuilder.arrayContains |
diff --git
a/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexReader.java
b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexReader.java
index 2abb7fb2cf..1dca420976 100644
---
a/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexReader.java
+++
b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexReader.java
@@ -63,6 +63,11 @@ public abstract class FileIndexReader implements
FunctionVisitor<FileIndexResult
return REMAIN;
}
+ @Override
+ public FileIndexResult visitArrayContains(FieldRef fieldRef, Object
literal) {
+ return REMAIN;
+ }
+
@Override
public FileIndexResult visitLike(FieldRef fieldRef, Object literal) {
return REMAIN;
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
index d30c758fd6..dd0c87b9b7 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/ConstantGlobalIndexReader.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/ConstantGlobalIndexReader.java
@@ -72,6 +72,12 @@ public class ConstantGlobalIndexReader implements
GlobalIndexReader {
return result;
}
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitArrayContains(
+ FieldRef fieldRef, Object literal) {
+ return result;
+ }
+
@Override
public CompletableFuture<Optional<GlobalIndexResult>> visitLike(
FieldRef fieldRef, Object literal) {
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexKeyExtractor.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexKeyExtractor.java
new file mode 100644
index 0000000000..9a0881fa13
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexKeyExtractor.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.globalindex;
+
+import org.apache.paimon.types.DataType;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.io.Serializable;
+
+/** Extracts zero or more normalized index keys from one source-column value.
*/
+public interface GlobalIndexKeyExtractor extends Serializable {
+
+ /** Type of the normalized keys emitted by {@link #extract(Object,
KeyConsumer)}. */
+ DataType keyType();
+
+ /** Emits normalized keys. Implementations may ignore null source values
and null elements. */
+ void extract(@Nullable Object sourceValue, KeyConsumer consumer) throws
IOException;
+
+ /** Whether extraction emits exactly the source value, including null. */
+ default boolean isIdentity() {
+ return false;
+ }
+
+ /** Creates an extractor which emits exactly one key for every source
value, including null. */
+ static GlobalIndexKeyExtractor identity(DataType keyType) {
+ return new IdentityKeyExtractor(keyType);
+ }
+
+ /** Identity extraction for scalar sorted indexes. */
+ class IdentityKeyExtractor implements GlobalIndexKeyExtractor {
+
+ private static final long serialVersionUID = 1L;
+
+ private final DataType keyType;
+
+ private IdentityKeyExtractor(DataType keyType) {
+ this.keyType = keyType;
+ }
+
+ @Override
+ public DataType keyType() {
+ return keyType;
+ }
+
+ @Override
+ public void extract(@Nullable Object sourceValue, KeyConsumer
consumer) throws IOException {
+ consumer.accept(sourceValue);
+ }
+
+ @Override
+ public boolean isIdentity() {
+ return true;
+ }
+ }
+
+ /** Consumer which lets an index build propagate storage failures from the
emitted keys. */
+ @FunctionalInterface
+ interface KeyConsumer {
+
+ void accept(@Nullable Object key) throws IOException;
+ }
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexReader.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexReader.java
index b857fdc1a1..065bd8814e 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexReader.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexReader.java
@@ -41,6 +41,12 @@ public interface GlobalIndexReader
return CompletableFuture.completedFuture(Optional.empty());
}
+ @Override
+ default CompletableFuture<Optional<GlobalIndexResult>> visitArrayContains(
+ FieldRef fieldRef, Object literal) {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+
@Override
default CompletableFuture<Optional<GlobalIndexResult>> visitBetween(
FieldRef fieldRef, Object from, Object to) {
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexSingleColumnWriter.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexSingleColumnWriter.java
index 67f5b95c07..301abbccd1 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexSingleColumnWriter.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexSingleColumnWriter.java
@@ -20,6 +20,8 @@ package org.apache.paimon.globalindex;
import javax.annotation.Nullable;
+import java.util.List;
+
/** Index writer for single-column global index with relative row id (from 0
to rowCnt - 1). */
public interface GlobalIndexSingleColumnWriter extends GlobalIndexWriter {
@@ -30,4 +32,14 @@ public interface GlobalIndexSingleColumnWriter extends
GlobalIndexWriter {
* @param relativeRowId local row id calculated by {@code rowId -
rangeStart}
*/
void write(@Nullable Object key, long relativeRowId);
+
+ /**
+ * Finishes a writer whose normalized entries cover {@code sourceRowCount}
source rows.
+ *
+ * <p>The default implementation preserves the one-entry-per-source-row
contract. Writers for
+ * zero-to-many key extraction should override this method.
+ */
+ default List<ResultEntry> finish(long sourceRowCount) {
+ return finish();
+ }
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/OffsetGlobalIndexReader.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/OffsetGlobalIndexReader.java
index 23d38d3a94..585984dcc0 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/OffsetGlobalIndexReader.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/OffsetGlobalIndexReader.java
@@ -74,6 +74,12 @@ public class OffsetGlobalIndexReader implements
GlobalIndexReader {
return wrapped.visitContains(fieldRef,
literal).thenApply(this::applyOffset);
}
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitArrayContains(
+ FieldRef fieldRef, Object literal) {
+ return wrapped.visitArrayContains(fieldRef,
literal).thenApply(this::applyOffset);
+ }
+
@Override
public CompletableFuture<Optional<GlobalIndexResult>> visitLike(
FieldRef fieldRef, Object literal) {
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/SortedFileMetaSelector.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/SortedFileMetaSelector.java
index 144b3195d1..b12ade7a2e 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/SortedFileMetaSelector.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/SortedFileMetaSelector.java
@@ -102,6 +102,11 @@ public class SortedFileMetaSelector implements
FunctionVisitor<Optional<List<Glo
return Optional.of(filter(meta -> literal != null &&
!meta.onlyNulls()));
}
+ @Override
+ public Optional<List<GlobalIndexIOMeta>> visitArrayContains(FieldRef
fieldRef, Object literal) {
+ return Optional.empty();
+ }
+
@Override
public Optional<List<GlobalIndexIOMeta>> visitLike(FieldRef fieldRef,
Object literal) {
return Optional.of(filter(meta -> literal != null &&
!meta.onlyNulls()));
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexSingleColumnWriter.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/SortedGlobalIndexer.java
similarity index 63%
copy from
paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexSingleColumnWriter.java
copy to
paimon-common/src/main/java/org/apache/paimon/globalindex/SortedGlobalIndexer.java
index 67f5b95c07..1f3bc4e603 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexSingleColumnWriter.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/SortedGlobalIndexer.java
@@ -18,16 +18,9 @@
package org.apache.paimon.globalindex;
-import javax.annotation.Nullable;
+/** A global indexer whose normalized keys must be sorted before they are
written. */
+public interface SortedGlobalIndexer extends GlobalIndexer {
-/** Index writer for single-column global index with relative row id (from 0
to rowCnt - 1). */
-public interface GlobalIndexSingleColumnWriter extends GlobalIndexWriter {
-
- /**
- * Write the indexed key and its related relative row id to the index file.
- *
- * @param key nullable index key
- * @param relativeRowId local row id calculated by {@code rowId -
rangeStart}
- */
- void write(@Nullable Object key, long relativeRowId);
+ /** Defines how source-column values are normalized into the keys consumed
by the writer. */
+ GlobalIndexKeyExtractor keyExtractor();
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/UnionGlobalIndexReader.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/UnionGlobalIndexReader.java
index 0e249b5df7..c2316b9427 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/UnionGlobalIndexReader.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/UnionGlobalIndexReader.java
@@ -77,6 +77,12 @@ public class UnionGlobalIndexReader implements
GlobalIndexReader {
return unionAsync(reader -> reader.visitContains(fieldRef, literal));
}
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitArrayContains(
+ FieldRef fieldRef, Object literal) {
+ return unionAsync(reader -> reader.visitArrayContains(fieldRef,
literal));
+ }
+
@Override
public CompletableFuture<Optional<GlobalIndexResult>> visitLike(
FieldRef fieldRef, Object literal) {
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexer.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexer.java
index e2a5c978f6..25ca2c0861 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexer.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexer.java
@@ -21,9 +21,11 @@ package org.apache.paimon.globalindex.bitmap;
import org.apache.paimon.compression.BlockCompressionFactory;
import org.apache.paimon.compression.CompressOptions;
import org.apache.paimon.globalindex.GlobalIndexIOMeta;
+import org.apache.paimon.globalindex.GlobalIndexKeyExtractor;
import org.apache.paimon.globalindex.GlobalIndexReader;
import org.apache.paimon.globalindex.GlobalIndexer;
import org.apache.paimon.globalindex.KeySerializer;
+import org.apache.paimon.globalindex.SortedGlobalIndexer;
import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
import org.apache.paimon.options.Options;
@@ -36,15 +38,17 @@ import java.util.List;
import java.util.concurrent.ExecutorService;
/** The {@link GlobalIndexer} for bitmap index. */
-public class BitmapGlobalIndexer implements GlobalIndexer {
+public class BitmapGlobalIndexer implements SortedGlobalIndexer {
private final KeySerializer keySerializer;
+ private final GlobalIndexKeyExtractor keyExtractor;
private final int dictionaryBlockSize;
@Nullable private final BlockCompressionFactory compressionFactory;
private final long fallbackScanMaxSize;
public BitmapGlobalIndexer(DataField dataField, Options options) {
this.keySerializer = KeySerializer.create(dataField.type());
+ this.keyExtractor = GlobalIndexKeyExtractor.identity(dataField.type());
this.dictionaryBlockSize =
(int)
options.get(BitmapGlobalIndexOptions.BITMAP_INDEX_DICTIONARY_BLOCK_SIZE)
@@ -59,6 +63,11 @@ public class BitmapGlobalIndexer implements GlobalIndexer {
.getBytes();
}
+ @Override
+ public GlobalIndexKeyExtractor keyExtractor() {
+ return keyExtractor;
+ }
+
@Override
public BitmapGlobalIndexWriter createWriter(GlobalIndexFileWriter
fileWriter)
throws IOException {
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/ConstantGlobalIndexReader.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/MultiValueBitmapIndexReader.java
similarity index 54%
copy from
paimon-common/src/main/java/org/apache/paimon/globalindex/ConstantGlobalIndexReader.java
copy to
paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/MultiValueBitmapIndexReader.java
index d30c758fd6..3305e3afa7 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/ConstantGlobalIndexReader.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/MultiValueBitmapIndexReader.java
@@ -16,128 +16,160 @@
* limitations under the License.
*/
-package org.apache.paimon.globalindex;
+package org.apache.paimon.globalindex.bitmap;
+import org.apache.paimon.globalindex.GlobalIndexIOMeta;
+import org.apache.paimon.globalindex.GlobalIndexReader;
+import org.apache.paimon.globalindex.GlobalIndexResult;
+import org.apache.paimon.globalindex.KeySerializer;
+import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
import org.apache.paimon.predicate.FieldRef;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataType;
+import java.io.IOException;
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));
+import java.util.concurrent.ExecutorService;
+
+/** Exposes array-element membership over the bitmap global index format. */
+public class MultiValueBitmapIndexReader implements GlobalIndexReader {
+
+ private final DataType elementType;
+ private final boolean compatibleElementType;
+ private final LazyFilteredBitmapReader bitmapReader;
+
+ MultiValueBitmapIndexReader(
+ GlobalIndexFileReader fileReader,
+ List<GlobalIndexIOMeta> files,
+ DataType elementType,
+ KeySerializer keySerializer,
+ long totalRowCount,
+ ExecutorService executor) {
+ this.elementType = elementType;
+ this.compatibleElementType =
+ files.stream()
+ .allMatch(
+ file ->
+
MultiValueIndexFileMeta.hasCompatibleElementType(
+ file.metadata(), elementType));
+ this.bitmapReader =
+ new LazyFilteredBitmapReader(
+ fileReader, files, keySerializer, 0, totalRowCount,
executor);
}
@Override
public CompletableFuture<Optional<GlobalIndexResult>>
visitIsNotNull(FieldRef fieldRef) {
- return result;
+ return unsupported();
}
@Override
public CompletableFuture<Optional<GlobalIndexResult>> visitIsNull(FieldRef
fieldRef) {
- return result;
+ return unsupported();
}
@Override
- public CompletableFuture<Optional<GlobalIndexResult>> visitIsNaN(FieldRef
fieldRef) {
- return result;
+ public CompletableFuture<Optional<GlobalIndexResult>> visitArrayContains(
+ FieldRef fieldRef, Object literal) {
+ if (!compatibleElementType
+ || !(fieldRef.type() instanceof ArrayType)
+ || !((ArrayType)
fieldRef.type()).getElementType().equals(elementType)) {
+ return unsupported();
+ }
+ return bitmapReader.visitEqual(fieldRef, literal);
}
@Override
public CompletableFuture<Optional<GlobalIndexResult>> visitStartsWith(
FieldRef fieldRef, Object literal) {
- return result;
+ return unsupported();
}
@Override
public CompletableFuture<Optional<GlobalIndexResult>> visitEndsWith(
FieldRef fieldRef, Object literal) {
- return result;
+ return unsupported();
}
@Override
public CompletableFuture<Optional<GlobalIndexResult>> visitContains(
FieldRef fieldRef, Object literal) {
- return result;
+ return unsupported();
}
@Override
public CompletableFuture<Optional<GlobalIndexResult>> visitLike(
FieldRef fieldRef, Object literal) {
- return result;
+ return unsupported();
}
@Override
public CompletableFuture<Optional<GlobalIndexResult>> visitLessThan(
FieldRef fieldRef, Object literal) {
- return result;
+ return unsupported();
}
@Override
public CompletableFuture<Optional<GlobalIndexResult>> visitGreaterOrEqual(
FieldRef fieldRef, Object literal) {
- return result;
+ return unsupported();
}
@Override
public CompletableFuture<Optional<GlobalIndexResult>> visitNotEqual(
FieldRef fieldRef, Object literal) {
- return result;
+ return unsupported();
}
@Override
public CompletableFuture<Optional<GlobalIndexResult>> visitLessOrEqual(
FieldRef fieldRef, Object literal) {
- return result;
+ return unsupported();
}
@Override
public CompletableFuture<Optional<GlobalIndexResult>> visitEqual(
FieldRef fieldRef, Object literal) {
- return result;
+ return unsupported();
}
@Override
public CompletableFuture<Optional<GlobalIndexResult>> visitGreaterThan(
FieldRef fieldRef, Object literal) {
- return result;
+ return unsupported();
}
@Override
public CompletableFuture<Optional<GlobalIndexResult>> visitIn(
FieldRef fieldRef, List<Object> literals) {
- return result;
+ return unsupported();
}
@Override
public CompletableFuture<Optional<GlobalIndexResult>> visitNotIn(
FieldRef fieldRef, List<Object> literals) {
- return result;
+ return unsupported();
}
@Override
public CompletableFuture<Optional<GlobalIndexResult>> visitBetween(
FieldRef fieldRef, Object from, Object to) {
- return result;
+ return unsupported();
}
@Override
public CompletableFuture<Optional<GlobalIndexResult>> visitNotBetween(
FieldRef fieldRef, Object from, Object to) {
- return result;
+ return unsupported();
}
@Override
- public void close() {}
+ public void close() throws IOException {
+ bitmapReader.close();
+ }
+
+ private static CompletableFuture<Optional<GlobalIndexResult>>
unsupported() {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/MultiValueBitmapIndexWriter.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/MultiValueBitmapIndexWriter.java
new file mode 100644
index 0000000000..f2c9982220
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/MultiValueBitmapIndexWriter.java
@@ -0,0 +1,167 @@
+/*
+ * 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.bitmap;
+
+import org.apache.paimon.compression.BlockCompressionFactory;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
+import org.apache.paimon.globalindex.KeySerializer;
+import org.apache.paimon.globalindex.ResultEntry;
+import org.apache.paimon.globalindex.SortedIndexFileMeta;
+import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
+import org.apache.paimon.memory.MemorySlice;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.utils.Preconditions;
+import org.apache.paimon.utils.RoaringNavigableMap64;
+
+import javax.annotation.Nullable;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+
+/** Streams one bitmap posting list per distinct non-null normalized array
element. */
+public class MultiValueBitmapIndexWriter implements
GlobalIndexSingleColumnWriter, Closeable {
+
+ private final GlobalIndexFileWriter fileWriter;
+ private final DataType elementType;
+ private final KeySerializer keySerializer;
+ private final Comparator<Object> comparator;
+ private final int dictionaryBlockSize;
+ @Nullable private final BlockCompressionFactory compressionFactory;
+ private final RoaringNavigableMap64 currentBitmap = new
RoaringNavigableMap64();
+
+ private String fileName;
+ private PositionOutputStream outputStream;
+ private BitmapGlobalIndexFormat.StreamingWriter streamingWriter;
+ private BitmapGlobalIndexFormat.SerializedKey currentKey;
+ private Object currentKeyObject;
+ private byte[] firstKey;
+ private byte[] lastKey;
+
+ MultiValueBitmapIndexWriter(
+ GlobalIndexFileWriter fileWriter,
+ DataType elementType,
+ KeySerializer keySerializer,
+ int dictionaryBlockSize,
+ @Nullable BlockCompressionFactory compressionFactory) {
+ this.fileWriter = fileWriter;
+ this.elementType = elementType;
+ this.keySerializer = keySerializer;
+ this.comparator = keySerializer.createComparator();
+ this.dictionaryBlockSize = dictionaryBlockSize;
+ this.compressionFactory = compressionFactory;
+ }
+
+ @Override
+ public void write(@Nullable Object key, long relativeRowId) {
+ Preconditions.checkArgument(
+ relativeRowId >= 0,
+ "Relative row id must be non-negative, but was %s.",
+ relativeRowId);
+ if (key == null) {
+ return;
+ }
+
+ if (currentKeyObject != null) {
+ int comparison = comparator.compare(key, currentKeyObject);
+ Preconditions.checkArgument(
+ comparison >= 0,
+ "Multivalue index keys must be written in monotonically
increasing order.");
+ if (comparison > 0) {
+ flushCurrentBitmap();
+ }
+ }
+ if (currentKeyObject == null) {
+ currentKey =
BitmapGlobalIndexFormat.SerializedKey.fromObject(keySerializer, key);
+ currentKeyObject =
keySerializer.deserialize(MemorySlice.wrap(currentKey.bytes()));
+ if (firstKey == null) {
+ firstKey = currentKey.bytes();
+ }
+ lastKey = currentKey.bytes();
+ }
+ currentBitmap.add(relativeRowId);
+ }
+
+ @Override
+ public List<ResultEntry> finish() {
+ throw new IllegalStateException(
+ "Multivalue index writers must be finished with the source row
count.");
+ }
+
+ @Override
+ public List<ResultEntry> finish(long sourceRowCount) {
+ Preconditions.checkArgument(
+ sourceRowCount >= 0,
+ "Source row count must be non-negative, but was %s.",
+ sourceRowCount);
+ if (sourceRowCount == 0) {
+ return Collections.emptyList();
+ }
+
+ try {
+ flushCurrentBitmap();
+ streamingWriter().finish(new RoaringNavigableMap64(), new
RoaringNavigableMap64());
+ close();
+ } catch (IOException e) {
+ throw new RuntimeException("Error in closing multivalue index
writer.", e);
+ }
+
+ byte[] meta =
+ MultiValueIndexFileMeta.serialize(
+ new SortedIndexFileMeta(firstKey, lastKey, false),
elementType);
+ return Collections.singletonList(new ResultEntry(fileName,
sourceRowCount, meta));
+ }
+
+ @Override
+ public void close() throws IOException {
+ PositionOutputStream stream = outputStream;
+ outputStream = null;
+ if (stream != null) {
+ stream.close();
+ }
+ }
+
+ private void flushCurrentBitmap() {
+ if (currentKey == null) {
+ return;
+ }
+ try {
+ streamingWriter().write(currentKey, currentBitmap);
+ currentBitmap.clear();
+ currentKey = null;
+ currentKeyObject = null;
+ } catch (IOException e) {
+ throw new RuntimeException("Error in writing multivalue index
files.", e);
+ }
+ }
+
+ private BitmapGlobalIndexFormat.StreamingWriter streamingWriter() throws
IOException {
+ if (streamingWriter == null) {
+ fileName =
fileWriter.newFileName(MultiValueGlobalIndexerFactory.IDENTIFIER);
+ outputStream = fileWriter.newOutputStream(fileName);
+ streamingWriter =
+ new BitmapGlobalIndexFormat.StreamingWriter(
+ outputStream, dictionaryBlockSize,
compressionFactory);
+ }
+ return streamingWriter;
+ }
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/MultiValueGlobalIndexOptions.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/MultiValueGlobalIndexOptions.java
new file mode 100644
index 0000000000..2cb146c605
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/MultiValueGlobalIndexOptions.java
@@ -0,0 +1,47 @@
+/*
+ * 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.bitmap;
+
+import org.apache.paimon.options.ConfigOption;
+import org.apache.paimon.options.ConfigOptions;
+import org.apache.paimon.options.MemorySize;
+
+/** Options for the bitmap-backed multivalue global index. */
+public class MultiValueGlobalIndexOptions {
+
+ public static final ConfigOption<MemorySize> DICTIONARY_BLOCK_SIZE =
+ ConfigOptions.key("multivalue-index.dictionary-block-size")
+ .memoryType()
+ .defaultValue(MemorySize.ofKibiBytes(16))
+ .withDescription("The target dictionary block size for
multivalue indexes.");
+
+ public static final ConfigOption<String> COMPRESSION =
+ ConfigOptions.key("multivalue-index.compression")
+ .stringType()
+ .defaultValue("none")
+ .withDescription("The compression algorithm for multivalue
dictionary blocks.");
+
+ public static final ConfigOption<Integer> COMPRESSION_LEVEL =
+ ConfigOptions.key("multivalue-index.compression-level")
+ .intType()
+ .defaultValue(1)
+ .withDescription("The compression level for multivalue
dictionary blocks.");
+
+ private MultiValueGlobalIndexOptions() {}
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/MultiValueGlobalIndexer.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/MultiValueGlobalIndexer.java
new file mode 100644
index 0000000000..e377206324
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/MultiValueGlobalIndexer.java
@@ -0,0 +1,127 @@
+/*
+ * 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.bitmap;
+
+import org.apache.paimon.compression.BlockCompressionFactory;
+import org.apache.paimon.compression.CompressOptions;
+import org.apache.paimon.data.InternalArray;
+import org.apache.paimon.globalindex.GlobalIndexIOMeta;
+import org.apache.paimon.globalindex.GlobalIndexKeyExtractor;
+import org.apache.paimon.globalindex.GlobalIndexReader;
+import org.apache.paimon.globalindex.KeySerializer;
+import org.apache.paimon.globalindex.SortedGlobalIndexer;
+import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
+import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.utils.Preconditions;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+
+/** Bitmap-backed multivalue index over the elements of an array column. */
+public class MultiValueGlobalIndexer implements SortedGlobalIndexer {
+
+ private final DataType elementType;
+ private final KeySerializer keySerializer;
+ private final GlobalIndexKeyExtractor keyExtractor;
+ private final int dictionaryBlockSize;
+ @Nullable private final BlockCompressionFactory compressionFactory;
+
+ public MultiValueGlobalIndexer(DataField dataField, Options options) {
+ Preconditions.checkArgument(
+ dataField.type() instanceof ArrayType,
+ "Multivalue index requires an ARRAY column, but column '%s'
has type %s.",
+ dataField.name(),
+ dataField.type());
+ this.elementType = ((ArrayType) dataField.type()).getElementType();
+ this.keySerializer = KeySerializer.create(elementType);
+ this.keyExtractor = new ArrayElementKeyExtractor(elementType);
+ this.dictionaryBlockSize =
+ (int)
options.get(MultiValueGlobalIndexOptions.DICTIONARY_BLOCK_SIZE).getBytes();
+ CompressOptions compressOptions =
+ new CompressOptions(
+ options.get(MultiValueGlobalIndexOptions.COMPRESSION),
+
options.get(MultiValueGlobalIndexOptions.COMPRESSION_LEVEL));
+ this.compressionFactory =
BlockCompressionFactory.create(compressOptions);
+ }
+
+ @Override
+ public GlobalIndexKeyExtractor keyExtractor() {
+ return keyExtractor;
+ }
+
+ @Override
+ public MultiValueBitmapIndexWriter createWriter(GlobalIndexFileWriter
fileWriter)
+ throws IOException {
+ return new MultiValueBitmapIndexWriter(
+ fileWriter, elementType, keySerializer, dictionaryBlockSize,
compressionFactory);
+ }
+
+ @Override
+ public GlobalIndexReader createReader(
+ GlobalIndexFileReader fileReader,
+ List<GlobalIndexIOMeta> files,
+ long totalRowCount,
+ ExecutorService executor) {
+ return new MultiValueBitmapIndexReader(
+ fileReader, files, elementType, keySerializer, totalRowCount,
executor);
+ }
+
+ private static class ArrayElementKeyExtractor implements
GlobalIndexKeyExtractor {
+
+ private static final long serialVersionUID = 1L;
+
+ private final DataType elementType;
+ private final InternalArray.ElementGetter elementGetter;
+
+ private ArrayElementKeyExtractor(DataType elementType) {
+ this.elementType = elementType;
+ this.elementGetter =
InternalArray.createElementGetter(elementType);
+ }
+
+ @Override
+ public DataType keyType() {
+ return elementType;
+ }
+
+ @Override
+ public void extract(@Nullable Object sourceValue, KeyConsumer
consumer) throws IOException {
+ if (sourceValue == null) {
+ return;
+ }
+ Preconditions.checkArgument(
+ sourceValue instanceof InternalArray,
+ "Multivalue index expects InternalArray values, but got
%s.",
+ sourceValue.getClass().getName());
+ InternalArray array = (InternalArray) sourceValue;
+ for (int i = 0; i < array.size(); i++) {
+ Object element = elementGetter.getElementOrNull(array, i);
+ if (element != null) {
+ consumer.accept(element);
+ }
+ }
+ }
+ }
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexSingleColumnWriter.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/MultiValueGlobalIndexerFactory.java
similarity index 54%
copy from
paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexSingleColumnWriter.java
copy to
paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/MultiValueGlobalIndexerFactory.java
index 67f5b95c07..dba47d9d3c 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexSingleColumnWriter.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/MultiValueGlobalIndexerFactory.java
@@ -16,18 +16,25 @@
* limitations under the License.
*/
-package org.apache.paimon.globalindex;
+package org.apache.paimon.globalindex.bitmap;
-import javax.annotation.Nullable;
+import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.GlobalIndexerFactory;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.DataField;
-/** Index writer for single-column global index with relative row id (from 0
to rowCnt - 1). */
-public interface GlobalIndexSingleColumnWriter extends GlobalIndexWriter {
+/** Factory for bitmap-backed multivalue indexes on array columns. */
+public class MultiValueGlobalIndexerFactory implements GlobalIndexerFactory {
- /**
- * Write the indexed key and its related relative row id to the index file.
- *
- * @param key nullable index key
- * @param relativeRowId local row id calculated by {@code rowId -
rangeStart}
- */
- void write(@Nullable Object key, long relativeRowId);
+ public static final String IDENTIFIER = "multivalue";
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public GlobalIndexer create(DataField indexField, Options options) {
+ return new MultiValueGlobalIndexer(indexField, options);
+ }
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/MultiValueIndexFileMeta.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/MultiValueIndexFileMeta.java
new file mode 100644
index 0000000000..08b4b45ad8
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/MultiValueIndexFileMeta.java
@@ -0,0 +1,73 @@
+/*
+ * 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.bitmap;
+
+import org.apache.paimon.globalindex.SortedIndexFileMeta;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.utils.JsonSerdeUtil;
+
+import javax.annotation.Nullable;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+
+/** Manifest-level sorted metadata with the element type used by a Multivalue
index. */
+public final class MultiValueIndexFileMeta {
+
+ // "MVIM". The trailer leaves the SortedIndexFileMeta prefix readable by
older selectors.
+ private static final int MAGIC = 0x4D56494D;
+ private static final int TRAILER_SIZE = Integer.BYTES * 2;
+
+ private MultiValueIndexFileMeta() {}
+
+ public static byte[] serialize(SortedIndexFileMeta sortedMeta, DataType
elementType) {
+ byte[] sortedBytes = sortedMeta.serialize();
+ byte[] typeBytes = typeSignature(elementType);
+ ByteBuffer buffer =
+ ByteBuffer.allocate(sortedBytes.length + typeBytes.length +
TRAILER_SIZE);
+ buffer.put(sortedBytes);
+ buffer.put(typeBytes);
+ buffer.putInt(typeBytes.length);
+ buffer.putInt(MAGIC);
+ return buffer.array();
+ }
+
+ public static boolean hasCompatibleElementType(
+ @Nullable byte[] metadata, DataType elementType) {
+ if (metadata == null || metadata.length < TRAILER_SIZE) {
+ return false;
+ }
+ ByteBuffer trailer =
+ ByteBuffer.wrap(metadata, metadata.length - TRAILER_SIZE,
TRAILER_SIZE);
+ int typeLength = trailer.getInt();
+ int magic = trailer.getInt();
+ if (magic != MAGIC || typeLength < 0 || typeLength > metadata.length -
TRAILER_SIZE) {
+ return false;
+ }
+ int typeOffset = metadata.length - TRAILER_SIZE - typeLength;
+ return Arrays.equals(
+ Arrays.copyOfRange(metadata, typeOffset, typeOffset +
typeLength),
+ typeSignature(elementType));
+ }
+
+ private static byte[] typeSignature(DataType elementType) {
+ return
JsonSerdeUtil.toJson(elementType).getBytes(StandardCharsets.UTF_8);
+ }
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java
index 07de8ff58c..660423c0cc 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java
@@ -21,9 +21,11 @@ package org.apache.paimon.globalindex.btree;
import org.apache.paimon.compression.BlockCompressionFactory;
import org.apache.paimon.compression.CompressOptions;
import org.apache.paimon.globalindex.GlobalIndexIOMeta;
+import org.apache.paimon.globalindex.GlobalIndexKeyExtractor;
import org.apache.paimon.globalindex.GlobalIndexReader;
import org.apache.paimon.globalindex.GlobalIndexer;
import org.apache.paimon.globalindex.KeySerializer;
+import org.apache.paimon.globalindex.SortedGlobalIndexer;
import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
import org.apache.paimon.io.cache.CacheManager;
@@ -58,15 +60,17 @@ import java.util.concurrent.ExecutorService;
*
* <p>This approach significantly reduces memory pressure during index reads.
*/
-public class BTreeGlobalIndexer implements GlobalIndexer {
+public class BTreeGlobalIndexer implements SortedGlobalIndexer {
private final KeySerializer keySerializer;
+ private final GlobalIndexKeyExtractor keyExtractor;
private final Options options;
private final long fallbackScanMaxSize;
private final LazyField<CacheManager> cacheManager;
public BTreeGlobalIndexer(DataField dataField, Options options) {
this.keySerializer = KeySerializer.create(dataField.type());
+ this.keyExtractor = GlobalIndexKeyExtractor.identity(dataField.type());
this.options = options;
this.fallbackScanMaxSize =
options.get(BTreeIndexOptions.BTREE_INDEX_FALLBACK_SCAN_MAX_SIZE).getBytes();
@@ -80,6 +84,11 @@ public class BTreeGlobalIndexer implements GlobalIndexer {
.BTREE_INDEX_HIGH_PRIORITY_POOL_RATIO)));
}
+ @Override
+ public GlobalIndexKeyExtractor keyExtractor() {
+ return keyExtractor;
+ }
+
@Override
public BTreeIndexWriter createWriter(GlobalIndexFileWriter fileWriter)
throws IOException {
long blockSize =
options.get(BTreeIndexOptions.BTREE_INDEX_BLOCK_SIZE).getBytes();
diff --git
a/paimon-common/src/main/java/org/apache/paimon/predicate/ArrayContains.java
b/paimon-common/src/main/java/org/apache/paimon/predicate/ArrayContains.java
new file mode 100644
index 0000000000..92f7f7b34a
--- /dev/null
+++ b/paimon-common/src/main/java/org/apache/paimon/predicate/ArrayContains.java
@@ -0,0 +1,90 @@
+/*
+ * 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.predicate;
+
+import org.apache.paimon.data.InternalArray;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.utils.Preconditions;
+
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+
+import java.util.List;
+import java.util.Optional;
+
+import static org.apache.paimon.predicate.CompareUtils.compareLiteral;
+
+/** A {@link LeafBinaryFunction} to test whether an array contains an element.
*/
+public class ArrayContains extends LeafBinaryFunction {
+
+ public static final String NAME = "ARRAY_CONTAINS";
+
+ public static final ArrayContains INSTANCE = new ArrayContains();
+
+ @JsonCreator
+ private ArrayContains() {}
+
+ @Override
+ public DataType literalType(DataType fieldType) {
+ return elementType(fieldType);
+ }
+
+ @Override
+ public boolean test(DataType type, Object field, Object literal) {
+ DataType elementType = elementType(type);
+ InternalArray array = (InternalArray) field;
+ InternalArray.ElementGetter getter =
InternalArray.createElementGetter(elementType);
+ for (int i = 0; i < array.size(); i++) {
+ Object element = getter.getElementOrNull(array, i);
+ if (element != null && compareLiteral(elementType, literal,
element) == 0) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public boolean test(
+ DataType type, long rowCount, Object min, Object max, Long
nullCount, Object literal) {
+ return true;
+ }
+
+ @Override
+ public Optional<LeafFunction> negate() {
+ return Optional.empty();
+ }
+
+ @Override
+ public <T> T visit(FunctionVisitor<T> visitor, FieldRef fieldRef,
List<Object> literals) {
+ return visitor.visitArrayContains(fieldRef, literals.get(0));
+ }
+
+ @Override
+ public String toJson() {
+ return NAME;
+ }
+
+ static DataType elementType(DataType fieldType) {
+ Preconditions.checkArgument(
+ fieldType instanceof ArrayType,
+ "ARRAY_CONTAINS requires an ARRAY field, but field type is
%s.",
+ fieldType);
+ return ((ArrayType) fieldType).getElementType();
+ }
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/predicate/FunctionVisitor.java
b/paimon-common/src/main/java/org/apache/paimon/predicate/FunctionVisitor.java
index f7040dae06..6216e1ce2d 100644
---
a/paimon-common/src/main/java/org/apache/paimon/predicate/FunctionVisitor.java
+++
b/paimon-common/src/main/java/org/apache/paimon/predicate/FunctionVisitor.java
@@ -66,6 +66,10 @@ public interface FunctionVisitor<T> extends
PredicateVisitor<T> {
T visitContains(FieldRef fieldRef, Object literal);
+ default T visitArrayContains(FieldRef fieldRef, Object literal) {
+ throw new UnsupportedOperationException();
+ }
+
T visitLike(FieldRef fieldRef, Object literal);
T visitLessThan(FieldRef fieldRef, Object literal);
diff --git
a/paimon-common/src/main/java/org/apache/paimon/predicate/LeafFunction.java
b/paimon-common/src/main/java/org/apache/paimon/predicate/LeafFunction.java
index 9ed1ba4338..b4b50695e0 100644
--- a/paimon-common/src/main/java/org/apache/paimon/predicate/LeafFunction.java
+++ b/paimon-common/src/main/java/org/apache/paimon/predicate/LeafFunction.java
@@ -63,6 +63,7 @@ public abstract class LeafFunction implements Serializable {
registry.put(StartsWith.NAME, StartsWith.INSTANCE);
registry.put(EndsWith.NAME, EndsWith.INSTANCE);
registry.put(Contains.NAME, Contains.INSTANCE);
+ registry.put(ArrayContains.NAME, ArrayContains.INSTANCE);
registry.put(Like.NAME, Like.INSTANCE);
registry.put(In.NAME, In.INSTANCE);
registry.put(NotIn.NAME, NotIn.INSTANCE);
@@ -89,6 +90,11 @@ public abstract class LeafFunction implements Serializable {
public abstract Optional<LeafFunction> negate();
+ /** Returns the type used to serialize literals for a field of the given
type. */
+ public DataType literalType(DataType fieldType) {
+ return fieldType;
+ }
+
@Override
public int hashCode() {
return this.getClass().getName().hashCode();
diff --git
a/paimon-common/src/main/java/org/apache/paimon/predicate/LeafPredicate.java
b/paimon-common/src/main/java/org/apache/paimon/predicate/LeafPredicate.java
index f68771bb7e..1383837f93 100644
--- a/paimon-common/src/main/java/org/apache/paimon/predicate/LeafPredicate.java
+++ b/paimon-common/src/main/java/org/apache/paimon/predicate/LeafPredicate.java
@@ -88,7 +88,8 @@ public class LeafPredicate implements Predicate {
@JsonProperty(FIELD_TRANSFORM) Transform transform,
@JsonProperty(FIELD_FUNCTION) LeafFunction function,
@JsonProperty(FIELD_LITERALS) List<Object> literals) {
- List<Object> convertedLiterals =
deserializeLiterals(transform.outputType(), literals);
+ List<Object> convertedLiterals =
+
deserializeLiterals(function.literalType(transform.outputType()), literals);
return new LeafPredicate(transform, function, convertedLiterals);
}
@@ -108,7 +109,7 @@ public class LeafPredicate implements Predicate {
@JsonGetter(FIELD_LITERALS)
public List<Object> literalsForJson() {
- return serializeLiterals(transform.outputType(), literals);
+ return serializeLiterals(function.literalType(transform.outputType()),
literals);
}
public List<String> fieldNames() {
@@ -208,7 +209,7 @@ public class LeafPredicate implements Predicate {
private ListSerializer<Object> literalsSerializer() {
return new ListSerializer<>(
NullableSerializer.wrapIfNullIsNotSupported(
- InternalSerializers.create(transform.outputType())));
+
InternalSerializers.create(function.literalType(transform.outputType()))));
}
private void writeObject(ObjectOutputStream out) throws IOException {
diff --git
a/paimon-common/src/main/java/org/apache/paimon/predicate/OnlyPartitionKeyEqualVisitor.java
b/paimon-common/src/main/java/org/apache/paimon/predicate/OnlyPartitionKeyEqualVisitor.java
index 677b2fb74f..35f482924d 100644
---
a/paimon-common/src/main/java/org/apache/paimon/predicate/OnlyPartitionKeyEqualVisitor.java
+++
b/paimon-common/src/main/java/org/apache/paimon/predicate/OnlyPartitionKeyEqualVisitor.java
@@ -67,6 +67,11 @@ public class OnlyPartitionKeyEqualVisitor implements
FunctionVisitor<Boolean> {
return false;
}
+ @Override
+ public Boolean visitArrayContains(FieldRef fieldRef, Object literal) {
+ return false;
+ }
+
@Override
public Boolean visitLike(FieldRef fieldRef, Object literal) {
return false;
diff --git
a/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateBuilder.java
b/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateBuilder.java
index b26baa2511..ed80a4f767 100644
---
a/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateBuilder.java
+++
b/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateBuilder.java
@@ -170,6 +170,17 @@ public class PredicateBuilder {
return leaf(Contains.INSTANCE, transform, patternLiteral);
}
+ public Predicate arrayContains(int idx, Object elementLiteral) {
+ DataField field = rowType.getFields().get(idx);
+ ArrayContains.elementType(field.type());
+ return leaf(ArrayContains.INSTANCE, idx, elementLiteral);
+ }
+
+ public Predicate arrayContains(Transform transform, Object elementLiteral)
{
+ ArrayContains.elementType(transform.outputType());
+ return leaf(ArrayContains.INSTANCE, transform, elementLiteral);
+ }
+
public Predicate like(int idx, Object patternLiteral) {
Pair<LeafBinaryFunction, Object> optimized =
LikeOptimization.tryOptimize(patternLiteral)
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 ec193fd5e9..8245da25f0 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
@@ -780,6 +780,91 @@ class GlobalIndexEvaluatorTest {
evaluator.close();
}
+ @Test
+ void testArrayContainsComposesAnyAndAllWithExistingBooleanEvaluation() {
+ RowType rowType =
+ new RowType(
+ Collections.singletonList(
+ new DataField(0, "tags",
DataTypes.ARRAY(DataTypes.INT()))));
+ GlobalIndexReader multiValueReader =
+ new StubGlobalIndexReader(null) {
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>>
visitArrayContains(
+ FieldRef fieldRef, Object literal) {
+ if (Integer.valueOf(1).equals(literal)) {
+ return
CompletableFuture.completedFuture(Optional.of(resultOf(1, 2)));
+ }
+ if (Integer.valueOf(2).equals(literal)) {
+ return
CompletableFuture.completedFuture(Optional.of(resultOf(2, 3)));
+ }
+ return CompletableFuture.completedFuture(
+ Optional.of(GlobalIndexResult.createEmpty()));
+ }
+ };
+ GlobalIndexEvaluator evaluator =
+ new GlobalIndexEvaluator(
+ rowType, fieldId ->
Collections.singletonList(multiValueReader));
+ PredicateBuilder builder = new PredicateBuilder(rowType);
+
+ Optional<GlobalIndexResult> any =
+ evaluator.evaluate(
+ PredicateBuilder.or(
+ builder.arrayContains(0, 1),
builder.arrayContains(0, 2)));
+ Optional<GlobalIndexResult> all =
+ evaluator.evaluate(
+ PredicateBuilder.and(
+ builder.arrayContains(0, 1),
builder.arrayContains(0, 2)));
+
+ assertThat(any).isPresent();
+ assertBitmapContainsExactly(any.get().results(), 1L, 2L, 3L);
+ assertThat(all).isPresent();
+ assertBitmapContainsExactly(all.get().results(), 2L);
+ evaluator.close();
+ }
+
+ @Test
+ void testArrayContainsFallbackAndReaderWrappers() {
+ RowType rowType =
+ new RowType(
+ Collections.singletonList(
+ new DataField(0, "tags",
DataTypes.ARRAY(DataTypes.INT()))));
+ Predicate predicate = new PredicateBuilder(rowType).arrayContains(0,
2);
+
+ GlobalIndexEvaluator unsupported =
+ new GlobalIndexEvaluator(
+ rowType,
+ fieldId -> Collections.singletonList(new
StubGlobalIndexReader(null)));
+ assertThat(unsupported.evaluate(predicate)).isEmpty();
+ unsupported.close();
+
+ GlobalIndexReader delegate =
+ new StubGlobalIndexReader(null) {
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>>
visitArrayContains(
+ FieldRef fieldRef, Object literal) {
+ return
CompletableFuture.completedFuture(Optional.of(resultOf(1, 3)));
+ }
+ };
+ GlobalIndexReader wrapped =
+ new UnionGlobalIndexReader(
+ Collections.singletonList(new
OffsetGlobalIndexReader(delegate, 10L, 20L)));
+ GlobalIndexEvaluator evaluator =
+ new GlobalIndexEvaluator(rowType, fieldId ->
Collections.singletonList(wrapped));
+
+ Optional<GlobalIndexResult> result = evaluator.evaluate(predicate);
+
+ assertThat(result).isPresent();
+ assertBitmapContainsExactly(result.get().results(), 11L, 13L);
+ Optional<GlobalIndexResult> constantResult =
+ new ConstantGlobalIndexReader(resultOf(4))
+ .visitArrayContains(
+ new FieldRef(0, "tags",
DataTypes.ARRAY(DataTypes.INT())), 2)
+ .join();
+ assertThat(constantResult).isPresent();
+ assertBitmapContainsExactly(constantResult.get().results(), 4L);
+ evaluator.close();
+ }
+
@Test
void testNotBetweenThroughUnionAndOffset() {
RowType rowType = rowType();
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/bitmap/MultiValueBitmapIndexReaderTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/bitmap/MultiValueBitmapIndexReaderTest.java
new file mode 100644
index 0000000000..98a5cba9f6
--- /dev/null
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/bitmap/MultiValueBitmapIndexReaderTest.java
@@ -0,0 +1,273 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.globalindex.bitmap;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.globalindex.GlobalIndexIOMeta;
+import org.apache.paimon.globalindex.GlobalIndexKeyExtractor;
+import org.apache.paimon.globalindex.GlobalIndexReader;
+import org.apache.paimon.globalindex.GlobalIndexResult;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
+import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.ResultEntry;
+import org.apache.paimon.globalindex.SortedGlobalIndexer;
+import org.apache.paimon.globalindex.SortedIndexFileMeta;
+import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
+import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.FieldRef;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+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.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.UUID;
+
+import static
org.apache.paimon.shade.guava30.com.google.common.util.concurrent.MoreExecutors.newDirectExecutorService;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for the bitmap-backed multivalue global index. */
+class MultiValueBitmapIndexReaderTest {
+
+ private final DataType arrayType = DataTypes.ARRAY(DataTypes.STRING());
+ private final DataField dataField = new DataField(1, "tags", arrayType);
+ private final FieldRef fieldRef = new FieldRef(1, "tags", arrayType);
+
+ private FileIO fileIO;
+ private Path basePath;
+ private GlobalIndexFileWriter fileWriter;
+ private GlobalIndexFileReader fileReader;
+ private GlobalIndexer globalIndexer;
+
+ @TempDir java.nio.file.Path tempPath;
+
+ @BeforeEach
+ void setUp() {
+ fileIO = LocalFileIO.create();
+ basePath = new Path(tempPath.toUri());
+ fileWriter =
+ new GlobalIndexFileWriter() {
+ @Override
+ public String newFileName(String prefix) {
+ return prefix + "-" + UUID.randomUUID() + ".index";
+ }
+
+ @Override
+ public PositionOutputStream newOutputStream(String
fileName)
+ throws IOException {
+ return fileIO.newOutputStream(new Path(basePath,
fileName), true);
+ }
+ };
+ fileReader = meta -> fileIO.newInputStream(meta.filePath());
+ globalIndexer = new MultiValueGlobalIndexer(dataField, new Options());
+ }
+
+ @Test
+ void testArrayContainsAndSafeFallback() throws Exception {
+ GlobalIndexSingleColumnWriter writer =
+ (GlobalIndexSingleColumnWriter)
globalIndexer.createWriter(fileWriter);
+ writer.write(str("A"), 0);
+ writer.write(str("A"), 0);
+ writer.write(str("B"), 0);
+ writer.write(str("B"), 3);
+ writer.write(str("C"), 4);
+
+ ResultEntry result = writer.finish(5).get(0);
+ assertThat(result.rowCount()).isEqualTo(5);
+ Path path = new Path(basePath, result.fileName());
+ GlobalIndexIOMeta meta =
+ new GlobalIndexIOMeta(path, fileIO.getFileSize(path),
result.meta());
+ assertThat(
+ MultiValueIndexFileMeta.hasCompatibleElementType(
+ result.meta(), DataTypes.STRING()))
+ .isTrue();
+ assertThat(
+ MultiValueIndexFileMeta.hasCompatibleElementType(
+ result.meta(), DataTypes.BIGINT()))
+ .isFalse();
+
+ try (GlobalIndexReader reader =
+ globalIndexer.createReader(
+ fileReader,
+ Collections.singletonList(meta),
+ 5,
+ newDirectExecutorService())) {
+ assertRows(reader.visitArrayContains(fieldRef, str("A")).join(),
0L);
+ assertRows(reader.visitArrayContains(fieldRef, str("B")).join(),
0L, 3L);
+ assertRows(reader.visitArrayContains(fieldRef,
str("missing")).join());
+ assertRows(reader.visitArrayContains(fieldRef, null).join());
+ assertThat(
+ reader.visitArrayContains(
+ new FieldRef(
+ 1, "tags",
DataTypes.ARRAY(DataTypes.BIGINT())),
+ 1L)
+ .join())
+ .isEmpty();
+ assertThat(reader.visitIsNull(fieldRef).join()).isEmpty();
+ assertThat(reader.visitIsNotNull(fieldRef).join()).isEmpty();
+
+ assertThat(reader.visitEqual(fieldRef,
array("A")).join()).isEmpty();
+ assertThat(reader.visitContains(fieldRef,
str("A")).join()).isEmpty();
+ }
+
+ GlobalIndexIOMeta legacyMeta =
+ new GlobalIndexIOMeta(
+ path,
+ fileIO.getFileSize(path),
+
SortedIndexFileMeta.deserialize(result.meta()).serialize());
+ try (GlobalIndexReader reader =
+ globalIndexer.createReader(
+ fileReader,
+ Collections.singletonList(legacyMeta),
+ 5,
+ newDirectExecutorService())) {
+ assertThat(reader.visitArrayContains(fieldRef,
str("A")).join()).isEmpty();
+ }
+ }
+
+ @Test
+ void testRowsWithoutIndexableElementsStillProduceAnIndex() throws
Exception {
+ GlobalIndexSingleColumnWriter writer =
+ (GlobalIndexSingleColumnWriter)
globalIndexer.createWriter(fileWriter);
+
+ ResultEntry result = writer.finish(3).get(0);
+ assertThat(result.rowCount()).isEqualTo(3);
+ Path path = new Path(basePath, result.fileName());
+ GlobalIndexIOMeta meta =
+ new GlobalIndexIOMeta(path, fileIO.getFileSize(path),
result.meta());
+ try (GlobalIndexReader reader =
+ globalIndexer.createReader(
+ fileReader,
+ Collections.singletonList(meta),
+ 3,
+ newDirectExecutorService())) {
+ assertRows(reader.visitArrayContains(fieldRef, str("A")).join());
+ }
+ }
+
+ @Test
+ void testNumericKeysUseLogicalOrder() throws Exception {
+ DataType intArrayType = DataTypes.ARRAY(DataTypes.INT());
+ DataField intField = new DataField(2, "numbers", intArrayType);
+ FieldRef intFieldRef = new FieldRef(2, "numbers", intArrayType);
+ GlobalIndexer intIndexer = new MultiValueGlobalIndexer(intField, new
Options());
+ GlobalIndexSingleColumnWriter writer =
+ (GlobalIndexSingleColumnWriter)
intIndexer.createWriter(fileWriter);
+ writer.write(-1, 0);
+ writer.write(0, 0);
+ writer.write(1, 1);
+
+ ResultEntry result = writer.finish(2).get(0);
+ Path path = new Path(basePath, result.fileName());
+ GlobalIndexIOMeta meta =
+ new GlobalIndexIOMeta(path, fileIO.getFileSize(path),
result.meta());
+ try (GlobalIndexReader reader =
+ intIndexer.createReader(
+ fileReader,
+ Collections.singletonList(meta),
+ 2,
+ newDirectExecutorService())) {
+ assertRows(reader.visitArrayContains(intFieldRef, -1).join(), 0L);
+ assertRows(reader.visitArrayContains(intFieldRef, 0).join(), 0L);
+ assertRows(reader.visitArrayContains(intFieldRef, 1).join(), 1L);
+ }
+ }
+
+ @Test
+ void testFactoryAndTypeValidation() throws Exception {
+ MultiValueGlobalIndexerFactory factory = new
MultiValueGlobalIndexerFactory();
+
assertThat(factory.identifier()).isEqualTo(MultiValueGlobalIndexerFactory.IDENTIFIER);
+ assertThat(factory.create(dataField, new Options()))
+ .isInstanceOf(MultiValueGlobalIndexer.class);
+ assertThat(globalIndexer).isInstanceOf(SortedGlobalIndexer.class);
+ GlobalIndexKeyExtractor extractor = ((SortedGlobalIndexer)
globalIndexer).keyExtractor();
+ assertThat(extractor.keyType()).isEqualTo(DataTypes.STRING());
+ List<Object> extracted = new ArrayList<>();
+ extractor.extract(array("B", null, "A"), extracted::add);
+ assertThat(extracted).containsExactly(str("B"), str("A"));
+ assertThatThrownBy(
+ () ->
+ new MultiValueGlobalIndexer(
+ new DataField(2, "scalar",
DataTypes.INT()), new Options()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("ARRAY");
+ assertThatThrownBy(
+ () ->
+ new MultiValueGlobalIndexer(
+ new DataField(
+ 3,
+ "nested",
+ DataTypes.ARRAY(
+ RowType.of(
+ new DataField(
+ 4,
+
"value",
+
DataTypes.INT())))),
+ new Options()))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("not supported by global index");
+ }
+
+ @Test
+ void testRejectsUnsortedNormalizedKeys() throws Exception {
+ GlobalIndexSingleColumnWriter writer =
+ (GlobalIndexSingleColumnWriter)
globalIndexer.createWriter(fileWriter);
+ writer.write(str("B"), 0);
+ assertThatThrownBy(() -> writer.write(str("A"), 1))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("monotonically increasing");
+ }
+
+ private static GenericArray array(String... values) {
+ BinaryString[] strings = new BinaryString[values.length];
+ for (int i = 0; i < values.length; i++) {
+ strings[i] = values[i] == null ? null : str(values[i]);
+ }
+ return new GenericArray(strings);
+ }
+
+ private static BinaryString str(String value) {
+ return BinaryString.fromString(value);
+ }
+
+ private static void assertRows(java.util.Optional<GlobalIndexResult>
result, Long... expected) {
+ assertThat(result).isPresent();
+ Iterator<Long> iterator = result.get().results().iterator();
+ List<Long> actual = new ArrayList<>();
+ while (iterator.hasNext()) {
+ actual.add(iterator.next());
+ }
+ assertThat(actual).containsExactlyInAnyOrder(expected);
+ }
+}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/predicate/LeafPredicateTest.java
b/paimon-common/src/test/java/org/apache/paimon/predicate/LeafPredicateTest.java
index d38703f7be..7fac8ddb88 100644
---
a/paimon-common/src/test/java/org/apache/paimon/predicate/LeafPredicateTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/predicate/LeafPredicateTest.java
@@ -71,6 +71,28 @@ class LeafPredicateTest {
assertThat(clone.toString()).isEqualTo(predicate.toString());
}
+ @Test
+ public void testArrayContainsSerializationUsesElementSerializer()
+ throws IOException, ClassNotFoundException {
+ PredicateBuilder builder =
+ new
PredicateBuilder(RowType.of(DataTypes.ARRAY(DataTypes.STRING())));
+ LeafPredicate predicate =
+ (LeafPredicate) builder.arrayContains(0,
BinaryString.fromString("vip"));
+
+ LeafPredicate clone = InstantiationUtil.clone(predicate);
+
+ assertThat(clone).isEqualTo(predicate);
+ assertThat(
+ clone.test(
+ GenericRow.of(
+ new GenericArray(
+ new BinaryString[] {
+
BinaryString.fromString("trial"),
+
BinaryString.fromString("vip")
+ }))))
+ .isTrue();
+ }
+
private LeafPredicate create() {
List<Object> inputs = new ArrayList<>();
inputs.add(new FieldRef(0, "f0", DataTypes.STRING()));
diff --git
a/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateBuilderTest.java
b/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateBuilderTest.java
index fe66f95241..b3f1c616b8 100644
---
a/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateBuilderTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateBuilderTest.java
@@ -18,9 +18,11 @@
package org.apache.paimon.predicate;
+import org.apache.paimon.data.GenericArray;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.format.SimpleColStats;
import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.DateType;
import org.apache.paimon.types.DecimalType;
import org.apache.paimon.types.IntType;
@@ -45,6 +47,7 @@ import java.util.stream.IntStream;
import static org.apache.paimon.predicate.SimpleColStatsTestUtils.test;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link PredicateBuilder}. */
public class PredicateBuilderTest {
@@ -220,6 +223,42 @@ public class PredicateBuilderTest {
assertThat(predicate.test(GenericRow.of(10))).isEqualTo(false);
}
+ @Test
+ public void testArrayContains() {
+ PredicateBuilder builder =
+ new
PredicateBuilder(RowType.of(DataTypes.ARRAY(DataTypes.INT())));
+ Predicate containsTwo = builder.arrayContains(0, 2);
+ Predicate transformedContainsTwo =
+ builder.arrayContains(
+ new FieldTransform(new FieldRef(0, "f0",
DataTypes.ARRAY(DataTypes.INT()))),
+ 2);
+
+ assertThat(containsTwo.test(GenericRow.of(new GenericArray(new
Integer[] {1, null, 2, 2}))))
+ .isTrue();
+ assertThat(
+ transformedContainsTwo.test(
+ GenericRow.of(new GenericArray(new Integer[]
{1, 2}))))
+ .isTrue();
+ assertThat(containsTwo.test(GenericRow.of(new GenericArray(new
Integer[] {1, 3}))))
+ .isFalse();
+ assertThat(containsTwo.test(GenericRow.of(new GenericArray(new
Integer[0])))).isFalse();
+ assertThat(containsTwo.test(GenericRow.of((Object) null))).isFalse();
+ assertThat(
+ builder.arrayContains(0, null)
+ .test(GenericRow.of(new GenericArray(new
Integer[] {1, null}))))
+ .isFalse();
+ assertThat(containsTwo.negate()).isEmpty();
+ }
+
+ @Test
+ public void testArrayContainsRequiresArrayField() {
+ PredicateBuilder builder = new
PredicateBuilder(RowType.of(DataTypes.INT()));
+
+ assertThatThrownBy(() -> builder.arrayContains(0, 1))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("ARRAY_CONTAINS requires an ARRAY
field");
+ }
+
// ---- or()/and() binary tree structure tests ----
@Test
diff --git
a/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateJsonSerdeTest.java
b/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateJsonSerdeTest.java
index 4837b9863f..e3814736c5 100644
---
a/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateJsonSerdeTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateJsonSerdeTest.java
@@ -171,6 +171,11 @@ class PredicateJsonSerdeTest {
.expectJson(
"{\"kind\":\"LEAF\",\"transform\":{\"name\":\"FIELD_REF\",\"fieldRef\":{\"index\":2,\"name\":\"f2\",\"type\":\"STRING\"}},\"function\":\"CONTAINS\",\"literals\":[\"foo\"]}"),
+ // LeafPredicate - ArrayContains uses the element type for
literal serde
+ TestSpec.forPredicate(builder.arrayContains(4,
BinaryString.fromString("vip")))
+ .expectJson(
+
"{\"kind\":\"LEAF\",\"transform\":{\"name\":\"FIELD_REF\",\"fieldRef\":{\"index\":4,\"name\":\"f4\",\"type\":{\"type\":\"ARRAY\",\"element\":\"STRING\"}}},\"function\":\"ARRAY_CONTAINS\",\"literals\":[\"vip\"]}"),
+
// LeafPredicate - Between
TestSpec.forPredicate(builder.between(0, 3, 7))
.expectJson(
@@ -249,7 +254,12 @@ class PredicateJsonSerdeTest {
private static PredicateBuilder newBuilder() {
return new PredicateBuilder(
- RowType.of(new IntType(), DataTypes.STRING(),
DataTypes.STRING(), new IntType()));
+ RowType.of(
+ new IntType(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ new IntType(),
+ DataTypes.ARRAY(DataTypes.STRING())));
}
private static List<Object> manyInts() {
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/orc/filter/OrcPredicateFunctionVisitor.java
b/paimon-format/src/main/java/org/apache/paimon/format/orc/filter/OrcPredicateFunctionVisitor.java
index d41611d999..b67364d33e 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/orc/filter/OrcPredicateFunctionVisitor.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/orc/filter/OrcPredicateFunctionVisitor.java
@@ -82,6 +82,11 @@ public class OrcPredicateFunctionVisitor
return Optional.empty();
}
+ @Override
+ public Optional<OrcFilters.Predicate> visitArrayContains(FieldRef
fieldRef, Object literal) {
+ return Optional.empty();
+ }
+
@Override
public Optional<OrcFilters.Predicate> visitLike(FieldRef fieldRef, Object
literal) {
return Optional.empty();
diff --git
a/paimon-format/src/main/java/org/apache/parquet/filter2/predicate/ParquetFilters.java
b/paimon-format/src/main/java/org/apache/parquet/filter2/predicate/ParquetFilters.java
index 645f82181a..9afb6f8981 100644
---
a/paimon-format/src/main/java/org/apache/parquet/filter2/predicate/ParquetFilters.java
+++
b/paimon-format/src/main/java/org/apache/parquet/filter2/predicate/ParquetFilters.java
@@ -199,6 +199,11 @@ public class ParquetFilters {
throw new UnsupportedOperationException();
}
+ @Override
+ public FilterPredicate visitArrayContains(FieldRef fieldRef, Object
literal) {
+ throw new UnsupportedOperationException();
+ }
+
@Override
public FilterPredicate visitLike(FieldRef fieldRef, Object literal) {
throw new UnsupportedOperationException();
