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 c07f35f271 [format][python] Support additional MAP BLOB key types
(#8963)
c07f35f271 is described below
commit c07f35f271bbcaa80a14d862bbfcc3fcc9613b61
Author: QuakeWang <[email protected]>
AuthorDate: Mon Aug 3 19:06:24 2026 +0800
[format][python] Support additional MAP BLOB key types (#8963)
---
docs/docs/concepts/spec/fileformat.md | 17 +++-
docs/docs/multimodal-table/blob.mdx | 4 +-
docs/docs/primary-key-table/blob-storage.md | 3 +-
.../test/java/org/apache/paimon/JavaPyE2ETest.java | 88 ++++++++++++++++--
.../format/blob/MapBlobElementSerializer.java | 78 ++++++++++++++++
.../paimon/format/blob/BlobFileFormatTest.java | 53 ++++++++++-
paimon-python/dev/run_mixed_tests.sh | 26 +++---
.../pypaimon/common/map_blob_key_serializer.py | 103 +++++++++++++++++++++
paimon-python/pypaimon/tests/blob_test.py | 64 +++++++++----
.../pypaimon/tests/e2e/java_py_read_write_test.py | 70 ++++++++++++++
10 files changed, 461 insertions(+), 45 deletions(-)
diff --git a/docs/docs/concepts/spec/fileformat.md
b/docs/docs/concepts/spec/fileformat.md
index 8c656874e1..7701184ac9 100644
--- a/docs/docs/concepts/spec/fileformat.md
+++ b/docs/docs/concepts/spec/fileformat.md
@@ -885,9 +885,20 @@ For `MAP<K, BLOB>`, the variable-length data area uses the
following nested payl
```
The key and Blob length indexes are aligned by entry position. A length of `-1`
-represents null, while zero represents an empty key or Blob. Supported key
types are
-the integer family, `CHAR`, and `VARCHAR`. An empty map has an entry count of
zero and
-is distinct from a null map.
+represents null, while zero represents an empty key or Blob. Supported key
types and
+their encodings are:
+
+| Key type | Encoding |
+|----------|----------|
+| `TINYINT`, `SMALLINT`, `INT`, `BIGINT` | Signed integer in little-endian
byte order using the type's fixed width |
+| `BOOLEAN` | One byte: `0` for false and `1` for true |
+| `DECIMAL(p, s)`, `p <= 18` | Eight-byte little-endian signed unscaled
integer |
+| `DECIMAL(p, s)`, `p > 18` | Minimal-length signed big-endian
two's-complement unscaled integer |
+| `DATE` | Four-byte little-endian signed count of days since 1970-01-01 |
+| `CHAR`, `VARCHAR` | UTF-8 bytes |
+
+The DECIMAL scale is defined by the field type and is not stored in each key.
An empty
+map has an entry count of zero and is distinct from a null map.
At the outer file index level, `-1` represents a null field and `-2`
represents a
field placeholder used by data evolution.
diff --git a/docs/docs/multimodal-table/blob.mdx
b/docs/docs/multimodal-table/blob.mdx
index 7d3f14d8c3..400b83766a 100644
--- a/docs/docs/multimodal-table/blob.mdx
+++ b/docs/docs/multimodal-table/blob.mdx
@@ -87,8 +87,8 @@ Paimon supports three storage modes for BLOB fields, selected
via **comment dire
This allows one table to mix different storage modes for different BLOB
columns.
`ARRAY<BLOB>` and `MAP<K, BLOB>` are supported only by `__BLOB_FIELD`;
descriptor-only and blob-view comment directives accept scalar BLOB fields
only.
-Map keys support the integer family, `CHAR`, and `VARCHAR`. Use non-null keys
for
-compatibility across Flink, Spark, and Python.
+Map keys support the integer family, `BOOLEAN`, `DECIMAL`, `DATE`, `CHAR`, and
+`VARCHAR`. Use non-null keys for compatibility across Flink, Spark, and Python.
## Table Options
diff --git a/docs/docs/primary-key-table/blob-storage.md
b/docs/docs/primary-key-table/blob-storage.md
index 51d7fc2756..7469d88044 100644
--- a/docs/docs/primary-key-table/blob-storage.md
+++ b/docs/docs/primary-key-table/blob-storage.md
@@ -94,7 +94,8 @@ array order, a null array, and null elements are preserved.
An empty array write
`MAP<K, BLOB>` is externalized value by value. Keys remain in the normal data
file and every non-null value is replaced
with a descriptor to managed storage. A null map, an empty map, and null
values are preserved. Supported key types are
-the integer family, `CHAR`, and `VARCHAR`; `blob-descriptor-field` and
`blob-view-field` remain scalar-only declarations.
+the integer family, `BOOLEAN`, `DECIMAL`, `DATE`, `CHAR`, and `VARCHAR`;
`blob-descriptor-field` and
+`blob-view-field` remain scalar-only declarations.
`blob.target-file-size` controls when a writer rolls to a new managed payload
pack. A pack can contain payloads from
multiple rows, and a row descriptor records its URI, offset, and length.
diff --git a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
index bdc963e678..403ec3c2dc 100644
--- a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
@@ -29,6 +29,7 @@ import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.BinaryVector;
import org.apache.paimon.data.BlobData;
import org.apache.paimon.data.DataFormatTestUtil;
+import org.apache.paimon.data.Decimal;
import org.apache.paimon.data.GenericArray;
import org.apache.paimon.data.GenericMap;
import org.apache.paimon.data.GenericRow;
@@ -92,6 +93,7 @@ import
org.junit.jupiter.api.condition.EnabledIfSystemProperty;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
@@ -1500,7 +1502,7 @@ public class JavaPyE2ETest {
assertThat(rows.get(4).get(0)).isEqualTo(lastValue.getBytes(StandardCharsets.UTF_8));
}
- /** Java writes a MAP<INT, BLOB> table for Python to read. */
+ /** Java writes MAP<K, BLOB> columns for Python to read. */
@Test
@EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
public void testJavaWriteMapBlobTable() throws Exception {
@@ -1510,6 +1512,16 @@ public class JavaPyE2ETest {
Schema.newBuilder()
.column("id", DataTypes.INT())
.column("payloads", DataTypes.MAP(DataTypes.INT(),
DataTypes.BLOB()))
+ .column(
+ "boolean_payloads",
+ DataTypes.MAP(DataTypes.BOOLEAN(),
DataTypes.BLOB()))
+ .column(
+ "compact_decimal_payloads",
+ DataTypes.MAP(DataTypes.DECIMAL(10, 2),
DataTypes.BLOB()))
+ .column(
+ "high_decimal_payloads",
+ DataTypes.MAP(DataTypes.DECIMAL(20, 2),
DataTypes.BLOB()))
+ .column("date_payloads",
DataTypes.MAP(DataTypes.DATE(), DataTypes.BLOB()))
.option(ROW_TRACKING_ENABLED.key(), "true")
.option(DATA_EVOLUTION_ENABLED.key(), "true")
.option(BUCKET.key(), "-1")
@@ -1522,28 +1534,51 @@ public class JavaPyE2ETest {
first.put(3, new BlobData(new byte[0]));
Map<Object, Object> last = new LinkedHashMap<>();
last.put(4, new
BlobData("java-omega".getBytes(StandardCharsets.UTF_8)));
+ Map<Object, Object> booleanPayloads = new LinkedHashMap<>();
+ booleanPayloads.put(true, new
BlobData("java-boolean".getBytes(StandardCharsets.UTF_8)));
+ Map<Object, Object> compactDecimalPayloads = new LinkedHashMap<>();
+ compactDecimalPayloads.put(
+ Decimal.fromBigDecimal(new BigDecimal("12.34"), 10, 2),
+ new
BlobData("java-compact-decimal".getBytes(StandardCharsets.UTF_8)));
+ Map<Object, Object> highDecimalPayloads = new LinkedHashMap<>();
+ highDecimalPayloads.put(
+ Decimal.fromBigDecimal(new
BigDecimal("123456789012345678.90"), 20, 2),
+ new
BlobData("java-high-decimal".getBytes(StandardCharsets.UTF_8)));
+ Map<Object, Object> datePayloads = new LinkedHashMap<>();
+ datePayloads.put(-1, new
BlobData("java-date".getBytes(StandardCharsets.UTF_8)));
FileStoreTable table = (FileStoreTable) catalog.getTable(identifier);
BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
try (BatchTableWrite write = writeBuilder.newWrite();
BatchTableCommit commit = writeBuilder.newCommit()) {
- write.write(GenericRow.of(1, new GenericMap(first)));
- write.write(GenericRow.of(2, new
GenericMap(Collections.emptyMap())));
- write.write(GenericRow.of(3, null));
- write.write(GenericRow.of(4, new GenericMap(last)));
+ write.write(
+ GenericRow.of(
+ 1,
+ new GenericMap(first),
+ new GenericMap(booleanPayloads),
+ new GenericMap(compactDecimalPayloads),
+ new GenericMap(highDecimalPayloads),
+ new GenericMap(datePayloads)));
+ write.write(
+ GenericRow.of(
+ 2, new GenericMap(Collections.emptyMap()), null,
null, null, null));
+ write.write(GenericRow.of(3, null, null, null, null, null));
+ write.write(GenericRow.of(4, new GenericMap(last), null, null,
null, null));
commit.commit(write.prepareCommit());
}
assertMapBlobRows(readMapBlobRows(table), "java-alpha", "java-omega");
+ assertAdditionalMapBlobKeyTypes(table, "java");
}
- /** Java reads a MAP<INT, BLOB> table written by Python. */
+ /** Java reads MAP<K, BLOB> columns written by Python. */
@Test
@EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
public void testJavaReadMapBlobTable() throws Exception {
FileStoreTable table =
(FileStoreTable)
catalog.getTable(identifier("map_blob_python_test"));
assertMapBlobRows(readMapBlobRows(table), "python-alpha",
"python-omega");
+ assertAdditionalMapBlobKeyTypes(table, "python");
}
private Map<Integer, Map<Integer, byte[]>> readMapBlobRows(FileStoreTable
table)
@@ -1588,6 +1623,47 @@ public class JavaPyE2ETest {
assertThat(rows.get(4).get(4)).isEqualTo(lastValue.getBytes(StandardCharsets.UTF_8));
}
+ private void assertAdditionalMapBlobKeyTypes(FileStoreTable table, String
valuePrefix)
+ throws Exception {
+ boolean[] found = new boolean[1];
+ List<Split> splits = new
ArrayList<>(table.newSnapshotReader().read().dataSplits());
+ try (org.apache.paimon.reader.RecordReader<InternalRow> reader =
+ table.newRead().createReader(splits)) {
+ reader.forEachRemaining(
+ row -> {
+ if (row.getInt(0) != 1) {
+ return;
+ }
+ found[0] = true;
+
+ InternalMap booleanMap = row.getMap(2);
+
assertThat(booleanMap.keyArray().getBoolean(0)).isTrue();
+ assertSingleBlobValue(booleanMap, valuePrefix +
"-boolean");
+
+ InternalMap compactDecimalMap = row.getMap(3);
+ assertThat(compactDecimalMap.keyArray().getDecimal(0,
10, 2).toBigDecimal())
+ .isEqualByComparingTo("12.34");
+ assertSingleBlobValue(compactDecimalMap, valuePrefix +
"-compact-decimal");
+
+ InternalMap highDecimalMap = row.getMap(4);
+ assertThat(highDecimalMap.keyArray().getDecimal(0, 20,
2).toBigDecimal())
+ .isEqualByComparingTo("123456789012345678.90");
+ assertSingleBlobValue(highDecimalMap, valuePrefix +
"-high-decimal");
+
+ InternalMap dateMap = row.getMap(5);
+ assertThat(dateMap.keyArray().getInt(0)).isEqualTo(-1);
+ assertSingleBlobValue(dateMap, valuePrefix + "-date");
+ });
+ }
+ assertThat(found[0]).isTrue();
+ }
+
+ private void assertSingleBlobValue(InternalMap map, String expectedValue) {
+ assertThat(map.size()).isOne();
+ assertThat(map.valueArray().getBlob(0).toData())
+ .isEqualTo(expectedValue.getBytes(StandardCharsets.UTF_8));
+ }
+
/** Java writes a VARIANT-column table for Python to read (Java→Python
E2E). */
@Test
@EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/blob/MapBlobElementSerializer.java
b/paimon-format/src/main/java/org/apache/paimon/format/blob/MapBlobElementSerializer.java
index 1a4b922552..3c67077e3d 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/blob/MapBlobElementSerializer.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/blob/MapBlobElementSerializer.java
@@ -24,6 +24,7 @@ import org.apache.paimon.data.BlobConsumer;
import org.apache.paimon.data.BlobDescriptor;
import org.apache.paimon.data.BlobFetchMetricReporter;
import org.apache.paimon.data.BlobMapPlaceholder;
+import org.apache.paimon.data.Decimal;
import org.apache.paimon.data.GenericMap;
import org.apache.paimon.data.InternalArray;
import org.apache.paimon.data.InternalMap;
@@ -33,6 +34,7 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.fs.SeekableInputStream;
import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DecimalType;
import org.apache.paimon.utils.DeltaVarintCompressor;
import org.apache.paimon.utils.IOUtils;
import org.apache.paimon.utils.Preconditions;
@@ -460,6 +462,13 @@ final class MapBlobElementSerializer implements
BlobElementSerializer {
return new IntKeySerializer();
case BIGINT:
return new BigIntKeySerializer();
+ case BOOLEAN:
+ return new BooleanKeySerializer();
+ case DECIMAL:
+ DecimalType decimalType = (DecimalType) keyType;
+ return new DecimalKeySerializer(decimalType.getPrecision(),
decimalType.getScale());
+ case DATE:
+ return new IntKeySerializer();
case CHAR:
case VARCHAR:
return new StringKeySerializer();
@@ -566,6 +575,75 @@ final class MapBlobElementSerializer implements
BlobElementSerializer {
}
}
+ /** {@link KeySerializer} for Boolean Type. */
+ private static final class BooleanKeySerializer implements KeySerializer {
+
+ @Override
+ public byte[] serialize(Object key) {
+ return new byte[] {(Boolean) key ? (byte) 1 : (byte) 0};
+ }
+
+ @Override
+ public Object deserialize(byte[] bytes) {
+ checkKeyLength(bytes, Byte.BYTES);
+ if (bytes[0] == 0) {
+ return false;
+ }
+ if (bytes[0] == 1) {
+ return true;
+ }
+ throw new IllegalArgumentException("Invalid MAP<X, BLOB> boolean
key.");
+ }
+
+ @Override
+ public int fixedLength() {
+ return Byte.BYTES;
+ }
+ }
+
+ /** {@link KeySerializer} for Decimal Type. */
+ private static final class DecimalKeySerializer implements KeySerializer {
+
+ private final int precision;
+ private final int scale;
+
+ private DecimalKeySerializer(int precision, int scale) {
+ this.precision = precision;
+ this.scale = scale;
+ }
+
+ @Override
+ public byte[] serialize(Object key) {
+ Decimal decimal = (Decimal) key;
+ return Decimal.isCompact(precision)
+ ? longToLittleEndian(decimal.toUnscaledLong())
+ : decimal.toUnscaledBytes();
+ }
+
+ @Override
+ public Object deserialize(byte[] bytes) {
+ Decimal decimal;
+ if (Decimal.isCompact(precision)) {
+ checkKeyLength(bytes, Long.BYTES);
+ decimal =
+ Decimal.fromUnscaledLong(
+ littleEndianBuffer(bytes).getLong(),
precision, scale);
+ } else {
+ decimal = Decimal.fromUnscaledBytes(bytes, precision, scale);
+ }
+ if (decimal == null) {
+ throw new IllegalArgumentException(
+ "MAP<X, BLOB> decimal key exceeds declared
precision.");
+ }
+ return decimal;
+ }
+
+ @Override
+ public int fixedLength() {
+ return Decimal.isCompact(precision) ? Long.BYTES : -1;
+ }
+ }
+
/** {@link KeySerializer} for String Type. */
private static final class StringKeySerializer implements KeySerializer {
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFileFormatTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFileFormatTest.java
index e77cb2a9d6..0c22894ee5 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFileFormatTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFileFormatTest.java
@@ -25,6 +25,7 @@ import org.apache.paimon.data.BlobData;
import org.apache.paimon.data.BlobMapPlaceholder;
import org.apache.paimon.data.BlobPlaceholder;
import org.apache.paimon.data.BlobRef;
+import org.apache.paimon.data.Decimal;
import org.apache.paimon.data.GenericArray;
import org.apache.paimon.data.GenericMap;
import org.apache.paimon.data.GenericRow;
@@ -51,6 +52,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.io.IOException;
+import java.math.BigDecimal;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
@@ -493,6 +495,10 @@ public class BlobFileFormatTest {
DataTypes.SMALLINT(),
DataTypes.INT(),
DataTypes.BIGINT(),
+ DataTypes.BOOLEAN(),
+ DataTypes.DECIMAL(10, 2),
+ DataTypes.DECIMAL(20, 2),
+ DataTypes.DATE(),
DataTypes.CHAR(10),
DataTypes.VARCHAR(10)
};
@@ -502,9 +508,36 @@ public class BlobFileFormatTest {
(short) 2,
3,
4L,
+ true,
+ Decimal.fromBigDecimal(new BigDecimal("12.34"), 10, 2),
+ Decimal.fromBigDecimal(new
BigDecimal("123456789012345678.90"), 20, 2),
+ -1,
BinaryString.fromString("char"),
BinaryString.fromString("varchar")
};
+ byte[][] serializedKeys =
+ new byte[][] {
+ {1},
+ {2, 0},
+ {3, 0, 0, 0},
+ {4, 0, 0, 0, 0, 0, 0, 0},
+ {1},
+ {(byte) 0xd2, 0x04, 0, 0, 0, 0, 0, 0},
+ {
+ 0,
+ (byte) 0xab,
+ 0x54,
+ (byte) 0xa9,
+ (byte) 0x8c,
+ (byte) 0xeb,
+ 0x1f,
+ 0x0a,
+ (byte) 0xd2
+ },
+ {(byte) 0xff, (byte) 0xff, (byte) 0xff, (byte) 0xff},
+ "char".getBytes(),
+ "varchar".getBytes()
+ };
for (int i = 0; i < keyTypes.length; i++) {
Path mapFile = new Path(parent, UUID.randomUUID().toString());
@@ -520,6 +553,24 @@ public class BlobFileFormatTest {
writer.close();
}
+ try (SeekableInputStream input = fileIO.newInputStream(mapFile)) {
+ BlobFileMeta fileMeta = new BlobFileMeta(input,
fileIO.getFileSize(mapFile), null);
+ int keyPosition =
+ Math.toIntExact(
+ fileMeta.blobOffset(0)
+ + Integer.BYTES
+ + Integer.BYTES
+ + Byte.BYTES
+ + Integer.BYTES);
+ byte[] fileBytes =
Files.readAllBytes(Paths.get(mapFile.toUri()));
+ assertThat(
+ Arrays.copyOfRange(
+ fileBytes,
+ keyPosition,
+ keyPosition +
serializedKeys[i].length))
+ .isEqualTo(serializedKeys[i]);
+ }
+
FormatReaderFactory readerFactory =
new BlobFileFormat(false,
BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE)
.createReaderFactory(null, rowType, null);
@@ -538,7 +589,7 @@ public class BlobFileFormatTest {
assertThatThrownBy(
() ->
BlobElementSerializerFactory.create(
- DataTypes.MAP(DataTypes.BOOLEAN(),
DataTypes.BLOB())))
+ DataTypes.MAP(DataTypes.FLOAT(),
DataTypes.BLOB())))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Unsupported key type");
assertThatThrownBy(
diff --git a/paimon-python/dev/run_mixed_tests.sh
b/paimon-python/dev/run_mixed_tests.sh
index 057e056591..d9371d4b41 100755
--- a/paimon-python/dev/run_mixed_tests.sh
+++ b/paimon-python/dev/run_mixed_tests.sh
@@ -975,40 +975,40 @@ run_array_blob_interop_test() {
}
run_map_blob_interop_test() {
- echo -e "${YELLOW}=== Running MAP<INT, BLOB> Test (Java Write → Python
Read, Python Write → Java Read) ===${NC}"
+ echo -e "${YELLOW}=== Running MAP<K, BLOB> Test (Java Write → Python Read,
Python Write → Java Read) ===${NC}"
if ! skip_batched_java_write; then
cd "$PROJECT_ROOT"
echo "Running Maven test for
JavaPyE2ETest.testJavaWriteMapBlobTable..."
if ! mvn test
-Dtest=org.apache.paimon.JavaPyE2ETest#testJavaWriteMapBlobTable -pl
paimon-core -q -Drun.e2e.tests=true; then
- echo -e "${RED}✗ Java MAP<INT, BLOB> write test failed${NC}"
+ echo -e "${RED}✗ Java MAP<K, BLOB> write test failed${NC}"
return 1
fi
- echo -e "${GREEN}✓ Java MAP<INT, BLOB> write test completed
successfully${NC}"
+ echo -e "${GREEN}✓ Java MAP<K, BLOB> write test completed
successfully${NC}"
fi
cd "$PAIMON_PYTHON_DIR"
- echo "Running Python MAP<INT, BLOB> read test..."
+ echo "Running Python MAP<K, BLOB> read test..."
if ! python -m pytest
java_py_read_write_test.py::JavaPyReadWriteTest::test_read_map_blob_written_by_java
-v; then
- echo -e "${RED}✗ Python MAP<INT, BLOB> read test failed${NC}"
+ echo -e "${RED}✗ Python MAP<K, BLOB> read test failed${NC}"
return 1
fi
- echo -e "${GREEN}✓ Python MAP<INT, BLOB> read test completed
successfully${NC}"
+ echo -e "${GREEN}✓ Python MAP<K, BLOB> read test completed
successfully${NC}"
- echo "Running Python MAP<INT, BLOB> write test..."
+ echo "Running Python MAP<K, BLOB> write test..."
if ! python -m pytest
java_py_read_write_test.py::JavaPyReadWriteTest::test_write_map_blob_for_java
-v; then
- echo -e "${RED}✗ Python MAP<INT, BLOB> write test failed${NC}"
+ echo -e "${RED}✗ Python MAP<K, BLOB> write test failed${NC}"
return 1
fi
- echo -e "${GREEN}✓ Python MAP<INT, BLOB> write test completed
successfully${NC}"
+ echo -e "${GREEN}✓ Python MAP<K, BLOB> write test completed
successfully${NC}"
cd "$PROJECT_ROOT"
echo "Running Maven test for JavaPyE2ETest.testJavaReadMapBlobTable..."
if ! mvn test
-Dtest=org.apache.paimon.JavaPyE2ETest#testJavaReadMapBlobTable -pl paimon-core
-q -Drun.e2e.tests=true; then
- echo -e "${RED}✗ Java MAP<INT, BLOB> read test failed${NC}"
+ echo -e "${RED}✗ Java MAP<K, BLOB> read test failed${NC}"
return 1
fi
- echo -e "${GREEN}✓ Java MAP<INT, BLOB> read test completed
successfully${NC}"
+ echo -e "${GREEN}✓ Java MAP<K, BLOB> read test completed successfully${NC}"
}
# Function to run VARIANT test (Java write, Python read)
@@ -1568,9 +1568,9 @@ main() {
fi
if [[ $map_blob_interop_result -eq 0 ]]; then
- echo -e "${GREEN}✓ MAP<INT, BLOB> Interoperability Test (Java ↔
Python): PASSED${NC}"
+ echo -e "${GREEN}✓ MAP<K, BLOB> Interoperability Test (Java ↔ Python):
PASSED${NC}"
else
- echo -e "${RED}✗ MAP<INT, BLOB> Interoperability Test (Java ↔ Python):
FAILED${NC}"
+ echo -e "${RED}✗ MAP<K, BLOB> Interoperability Test (Java ↔ Python):
FAILED${NC}"
fi
if [[ $data_evolution_result -eq 0 ]]; then
diff --git a/paimon-python/pypaimon/common/map_blob_key_serializer.py
b/paimon-python/pypaimon/common/map_blob_key_serializer.py
index 7617d4758c..51bc3123c6 100644
--- a/paimon-python/pypaimon/common/map_blob_key_serializer.py
+++ b/paimon-python/pypaimon/common/map_blob_key_serializer.py
@@ -15,11 +15,16 @@
# specific language governing permissions and limitations
# under the License.
+import datetime
import struct
+from decimal import Decimal as BigDecimal
from typing import Optional
+from pypaimon.data.decimal import Decimal
from pypaimon.schema.data_types import AtomicType, DataType
+_EPOCH_DATE = datetime.date(1970, 1, 1)
+
class MapBlobKeySerializer:
@@ -61,6 +66,97 @@ class MapBlobKeySerializer:
return struct.unpack(self._struct_format, data)[0]
+class BooleanMapBlobKeySerializer(MapBlobKeySerializer):
+
+ def __init__(self):
+ self.fixed_length = 1
+
+ def serialize(self, key) -> bytes:
+ if not isinstance(key, bool):
+ raise ValueError("MAP<X, BLOB> BOOLEAN key must be a boolean.")
+ return b'\x01' if key else b'\x00'
+
+ def deserialize(self, data: bytes):
+ if len(data) != self.fixed_length:
+ raise ValueError(
+ f"Expected {self.fixed_length} key bytes, but found
{len(data)}."
+ )
+ if data[0] == 0:
+ return False
+ if data[0] == 1:
+ return True
+ raise ValueError("Invalid MAP<X, BLOB> boolean key.")
+
+
+class DecimalMapBlobKeySerializer(MapBlobKeySerializer):
+
+ def __init__(self, type_name: str, precision: int, scale: int):
+ self._type_name = type_name
+ self._precision = precision
+ self._scale = scale
+ self.fixed_length = 8 if Decimal.is_compact_precision(precision) else
-1
+
+ def serialize(self, key) -> bytes:
+ if not isinstance(key, BigDecimal):
+ raise ValueError(
+ f"MAP<X, BLOB> {self._type_name} key must be a
decimal.Decimal."
+ )
+ decimal = Decimal.from_big_decimal(key, self._precision, self._scale)
+ if decimal is None:
+ raise ValueError(
+ f"MAP<X, BLOB> {self._type_name} key exceeds declared
precision."
+ )
+ if decimal.is_compact():
+ return struct.pack('<q', decimal.to_unscaled_long())
+ return decimal.to_unscaled_bytes()
+
+ def deserialize(self, data: bytes):
+ if self.fixed_length >= 0:
+ if len(data) != self.fixed_length:
+ raise ValueError(
+ f"Expected {self.fixed_length} key bytes, but found
{len(data)}."
+ )
+ decimal = Decimal.from_unscaled_long(
+ struct.unpack('<q', data)[0],
+ self._precision,
+ self._scale,
+ )
+ else:
+ if not data:
+ raise ValueError("Invalid MAP<X, BLOB> decimal key.")
+ decimal = Decimal.from_unscaled_bytes(
+ data,
+ self._precision,
+ self._scale,
+ )
+ if decimal is None:
+ raise ValueError(
+ "MAP<X, BLOB> decimal key exceeds declared precision."
+ )
+ return decimal.to_big_decimal()
+
+
+class DateMapBlobKeySerializer(MapBlobKeySerializer):
+
+ def __init__(self):
+ self.fixed_length = 4
+
+ def serialize(self, key) -> bytes:
+ if isinstance(key, datetime.datetime) or not isinstance(key,
datetime.date):
+ raise ValueError("MAP<X, BLOB> DATE key must be a datetime.date.")
+ try:
+ return struct.pack('<i', (key - _EPOCH_DATE).days)
+ except struct.error as error:
+ raise ValueError(f"MAP<X, BLOB> DATE key is out of range: {key}.")
from error
+
+ def deserialize(self, data: bytes):
+ if len(data) != self.fixed_length:
+ raise ValueError(
+ f"Expected {self.fixed_length} key bytes, but found
{len(data)}."
+ )
+ return _EPOCH_DATE + datetime.timedelta(days=struct.unpack('<i',
data)[0])
+
+
def create_map_blob_key_serializer(data_type: DataType) ->
MapBlobKeySerializer:
if not isinstance(data_type, AtomicType):
raise ValueError(f"Unsupported key type for MAP<X, BLOB>: {data_type}")
@@ -74,6 +170,13 @@ def create_map_blob_key_serializer(data_type: DataType) ->
MapBlobKeySerializer:
return MapBlobKeySerializer(type_name, '<i')
if type_name == 'BIGINT':
return MapBlobKeySerializer(type_name, '<q')
+ if type_name == 'BOOLEAN':
+ return BooleanMapBlobKeySerializer()
+ if type_name.startswith('DECIMAL'):
+ precision, scale = Decimal.extract_decimal_precision_scale(type_name)
+ return DecimalMapBlobKeySerializer(type_name, precision, scale)
+ if type_name == 'DATE':
+ return DateMapBlobKeySerializer()
if type_name == 'STRING' or type_name.startswith('CHAR') or
type_name.startswith('VARCHAR'):
return MapBlobKeySerializer(type_name)
raise ValueError(f"Unsupported key type for MAP<X, BLOB>: {data_type}")
diff --git a/paimon-python/pypaimon/tests/blob_test.py
b/paimon-python/pypaimon/tests/blob_test.py
index 71eb26db9c..e7dfc9d179 100644
--- a/paimon-python/pypaimon/tests/blob_test.py
+++ b/paimon-python/pypaimon/tests/blob_test.py
@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
+import datetime
import io
import os
import shutil
@@ -22,6 +23,7 @@ import struct
import tempfile
import unittest
import zlib
+from decimal import Decimal
from pathlib import Path
from unittest.mock import patch
@@ -3049,10 +3051,31 @@ class BlobEndToEndTest(unittest.TestCase):
(AtomicType("INT"), -2147483648),
(AtomicType("INTEGER"), 0x01020304),
(AtomicType("BIGINT"), -9223372036854775808),
+ (AtomicType("BOOLEAN"), True),
+ (AtomicType("DECIMAL(10, 2)"), Decimal("12.34")),
+ (
+ AtomicType("DECIMAL(20, 2)"),
+ Decimal("123456789012345678.90"),
+ ),
+ (AtomicType("DATE"), datetime.date(1969, 12, 31)),
(AtomicType("STRING"), "string"),
(AtomicType("CHAR(3)"), "abc"),
(AtomicType("VARCHAR(10)"), "varchar"),
]
+ serialized_keys = [
+ b"\x80",
+ b"\x00\x80",
+ b"\x00\x00\x00\x80",
+ b"\x04\x03\x02\x01",
+ b"\x00\x00\x00\x00\x00\x00\x00\x80",
+ b"\x01",
+ b"\xd2\x04\x00\x00\x00\x00\x00\x00",
+ b"\x00\xab\x54\xa9\x8c\xeb\x1f\x0a\xd2",
+ b"\xff\xff\xff\xff",
+ b"string",
+ b"abc",
+ b"varchar",
+ ]
for index, (key_type, key) in enumerate(cases):
with self.subTest(key_type=key_type):
blob_file_path = os.path.join(self.temp_dir,
f"map_key_{index}.blob")
@@ -3079,35 +3102,38 @@ class BlobEndToEndTest(unittest.TestCase):
result = next(iterator).values[0]
self.assertEqual(result[key].to_data(), b"value")
- if key_type.type == "INTEGER":
- with open(blob_file_path, 'rb') as blob_file:
- blob_file.seek(BlobRecordIterator.MAGIC_NUMBER_SIZE
- + BlobRecordIterator.MAP_HEADER_SIZE)
- self.assertEqual(blob_file.read(4),
b"\x04\x03\x02\x01")
- reader = FormatBlobReader(
- file_io=file_io,
- file_path=blob_file_path,
- read_fields=["blob_map"],
- full_fields=fields,
- push_down_predicate=None,
- blob_as_descriptor=False,
+ with open(blob_file_path, 'rb') as blob_file:
+ blob_file.seek(BlobRecordIterator.MAGIC_NUMBER_SIZE
+ + BlobRecordIterator.MAP_HEADER_SIZE)
+ self.assertEqual(
+ blob_file.read(len(serialized_keys[index])),
+ serialized_keys[index],
)
- try:
- value =
dict(reader.read_arrow_batch().column(0)[0].as_py())
- self.assertEqual(value, {key: b"value"})
- finally:
- reader.close()
+
+ reader = FormatBlobReader(
+ file_io=file_io,
+ file_path=blob_file_path,
+ read_fields=["blob_map"],
+ full_fields=fields,
+ push_down_predicate=None,
+ blob_as_descriptor=False,
+ )
+ try:
+ value =
dict(reader.read_arrow_batch().column(0)[0].as_py())
+ self.assertEqual(value, {key: b"value"})
+ finally:
+ reader.close()
output = io.BytesIO()
unsupported_key_writer = BlobFormatWriter(output)
unsupported_key_field = DataField(
0,
"blob_map",
- MapType(True, AtomicType("BOOLEAN"), AtomicType("BLOB")),
+ MapType(True, AtomicType("FLOAT"), AtomicType("BLOB")),
)
with self.assertRaisesRegex(ValueError, "Unsupported key type"):
unsupported_key_writer.add_element(GenericRow(
- [{True: BlobData(b"value")}],
+ [{1.0: BlobData(b"value")}],
[unsupported_key_field],
RowKind.INSERT,
))
diff --git a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
index 33bfe83e19..5db6840a31 100644
--- a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
+++ b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
@@ -20,6 +20,7 @@ import json
import os
import sys
import unittest
+from decimal import Decimal
import pandas as pd
import pyarrow as pa
@@ -1552,12 +1553,40 @@ class JavaPyReadWriteTest(unittest.TestCase):
{4: b'java-omega'},
],
)
+ expected_additional_payloads = {
+ 'boolean_payloads': {True: b'java-boolean'},
+ 'compact_decimal_payloads': {
+ Decimal('12.34'): b'java-compact-decimal',
+ },
+ 'high_decimal_payloads': {
+ Decimal('123456789012345678.90'): b'java-high-decimal',
+ },
+ 'date_payloads': {
+ datetime.date(1969, 12, 31): b'java-date',
+ },
+ }
+ for name, expected in expected_additional_payloads.items():
+ self.assertEqual(
+ [None if value is None else dict(value)
+ for value in result.column(name).to_pylist()],
+ [expected, None, None, None],
+ )
def test_write_map_blob_for_java(self):
map_blob_type = pa.map_(pa.int32(), pa.large_binary())
+ boolean_map_blob_type = pa.map_(pa.bool_(), pa.large_binary())
+ compact_decimal_map_blob_type = pa.map_(
+ pa.decimal128(10, 2), pa.large_binary())
+ high_decimal_map_blob_type = pa.map_(
+ pa.decimal128(20, 2), pa.large_binary())
+ date_map_blob_type = pa.map_(pa.date32(), pa.large_binary())
pa_schema = pa.schema([
('id', pa.int32()),
('payloads', map_blob_type),
+ ('boolean_payloads', boolean_map_blob_type),
+ ('compact_decimal_payloads', compact_decimal_map_blob_type),
+ ('high_decimal_payloads', high_decimal_map_blob_type),
+ ('date_payloads', date_map_blob_type),
])
schema = Schema.from_pyarrow_schema(
pa_schema,
@@ -1583,6 +1612,29 @@ class JavaPyReadWriteTest(unittest.TestCase):
],
type=map_blob_type,
),
+ 'boolean_payloads': pa.array(
+ [[(True, b'python-boolean')], None, None, None],
+ type=boolean_map_blob_type,
+ ),
+ 'compact_decimal_payloads': pa.array(
+ [[(Decimal('12.34'), b'python-compact-decimal')],
+ None, None, None],
+ type=compact_decimal_map_blob_type,
+ ),
+ 'high_decimal_payloads': pa.array(
+ [[(
+ Decimal('123456789012345678.90'),
+ b'python-high-decimal',
+ )], None, None, None],
+ type=high_decimal_map_blob_type,
+ ),
+ 'date_payloads': pa.array(
+ [[(
+ datetime.date(1969, 12, 31),
+ b'python-date',
+ )], None, None, None],
+ type=date_map_blob_type,
+ ),
}, schema=pa_schema)
write_builder = table.new_batch_write_builder()
table_write = write_builder.new_write()
@@ -1607,6 +1659,24 @@ class JavaPyReadWriteTest(unittest.TestCase):
{4: b'python-omega'},
],
)
+ expected_additional_payloads = {
+ 'boolean_payloads': {True: b'python-boolean'},
+ 'compact_decimal_payloads': {
+ Decimal('12.34'): b'python-compact-decimal',
+ },
+ 'high_decimal_payloads': {
+ Decimal('123456789012345678.90'): b'python-high-decimal',
+ },
+ 'date_payloads': {
+ datetime.date(1969, 12, 31): b'python-date',
+ },
+ }
+ for name, expected in expected_additional_payloads.items():
+ self.assertEqual(
+ [None if value is None else dict(value)
+ for value in result.column(name).to_pylist()],
+ [expected, None, None, None],
+ )
def test_compact_conflict_shard_update(self):
"""
