diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/CompletedFetch.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/CompletedFetch.java index de9cb1d3f6..b756ad94d5 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/CompletedFetch.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/CompletedFetch.java @@ -30,6 +30,7 @@ import org.apache.fluss.record.LogRecord; import org.apache.fluss.record.LogRecordBatch; import org.apache.fluss.record.LogRecordReadContext; +import org.apache.fluss.row.ColumnValueDecodingRow; import org.apache.fluss.row.GenericRow; import org.apache.fluss.row.InternalRow; import org.apache.fluss.rpc.protocol.ApiError; @@ -134,7 +135,7 @@ ScanRecord toScanRecord(LogRecord record) { record.logOffset(), record.timestamp(), record.getChangeType(), - newRow, + ColumnValueDecodingRow.wrap(newRow), getRecordSizeInBytes(record)); } diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/writer/MultiTableWriterImpl.java b/fluss-client/src/main/java/org/apache/fluss/client/table/writer/MultiTableWriterImpl.java index 5702188d3a..dd1f0ee4df 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/table/writer/MultiTableWriterImpl.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/table/writer/MultiTableWriterImpl.java @@ -443,7 +443,11 @@ private static final class PrimaryTableWriteState extends TableWriteState { tableInfo.getTableConfig().getDataLakeFormat().orElse(null)); this.kvFormat = tableInfo.getTableConfig().getKvFormat(); this.kvWriteFormat = WriteFormat.fromKvFormat(this.kvFormat); - this.kvRowEncoder = RowEncoder.create(this.kvFormat, rowType); + this.kvRowEncoder = + RowEncoder.create( + this.kvFormat, + tableInfo.getSchema(), + tableInfo.getProperties().toMap()); } @Override @@ -499,9 +503,13 @@ private byte[] encodeBucketKey(InternalRow row, byte[] primaryKey) { } private BinaryRow encodeKvRow(InternalRow row) { - if (kvFormat == KvFormat.INDEXED && row instanceof IndexedRow) { + if (!kvRowEncoder.hasColumnValueEncoding() + && kvFormat == KvFormat.INDEXED + && row instanceof IndexedRow) { return (IndexedRow) row; - } else if (kvFormat == KvFormat.COMPACTED && row instanceof CompactedRow) { + } else if (!kvRowEncoder.hasColumnValueEncoding() + && kvFormat == KvFormat.COMPACTED + && row instanceof CompactedRow) { return (CompactedRow) row; } kvRowEncoder.startNewRow(); diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/writer/UpsertWriterImpl.java b/fluss-client/src/main/java/org/apache/fluss/client/table/writer/UpsertWriterImpl.java index 8417855ccb..675b9f6afe 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/table/writer/UpsertWriterImpl.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/table/writer/UpsertWriterImpl.java @@ -97,7 +97,9 @@ class UpsertWriterImpl extends AbstractTableWriter implements UpsertWriter { this.kvFormat = tableInfo.getTableConfig().getKvFormat(); this.writeFormat = WriteFormat.fromKvFormat(this.kvFormat); - this.rowEncoder = RowEncoder.create(kvFormat, rowType); + this.rowEncoder = + RowEncoder.create( + kvFormat, tableInfo.getSchema(), tableInfo.getProperties().toMap()); this.fieldGetters = InternalRow.createFieldGetters(rowType); this.tableInfo = tableInfo; @@ -217,10 +219,14 @@ public CompletableFuture delete(InternalRow row) { return sendWithResult(record, DeleteResult::new); } - private BinaryRow encodeRow(InternalRow row) { - if (kvFormat == KvFormat.INDEXED && row instanceof IndexedRow) { + BinaryRow encodeRow(InternalRow row) { + if (!rowEncoder.hasColumnValueEncoding() + && kvFormat == KvFormat.INDEXED + && row instanceof IndexedRow) { return (IndexedRow) row; - } else if (kvFormat == KvFormat.COMPACTED && row instanceof CompactedRow) { + } else if (!rowEncoder.hasColumnValueEncoding() + && kvFormat == KvFormat.COMPACTED + && row instanceof CompactedRow) { return (CompactedRow) row; } diff --git a/fluss-common/src/main/java/org/apache/fluss/compression/ColumnValueCodec.java b/fluss-common/src/main/java/org/apache/fluss/compression/ColumnValueCodec.java new file mode 100644 index 0000000000..c5cf04b2b7 --- /dev/null +++ b/fluss-common/src/main/java/org/apache/fluss/compression/ColumnValueCodec.java @@ -0,0 +1,319 @@ +/* + * 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.fluss.compression; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.exception.CorruptRecordException; +import org.apache.fluss.exception.FlussRuntimeException; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.types.DataTypeRoot; +import org.apache.fluss.utils.IOUtils; + +import com.github.luben.zstd.ZstdInputStream; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.BitSet; +import java.util.Collections; +import java.util.HashMap; +import java.util.Locale; +import java.util.Map; + +import static org.apache.fluss.config.FlussConfigUtils.getColumnCompressionConfigKey; +import static org.apache.fluss.utils.Preconditions.checkArgument; +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** + * Codec for self-describing values in top-level BYTES columns. + * + *

Each encoded value contains all information required for decoding: + * + * 

+ * +----------+---------+----------+-----------+---------+
+ * | Magic 4B | Ver. 1B | Codec 1B | Length 4B | Payload |
+ * +----------+---------+----------+-----------+---------+
+ *
+ * + *

The magic is {@code FCMP}, the codec identifies RAW, LZ4, or ZSTD, and the big-endian length + * is the exact decoded size. Values without the magic are legacy unwrapped values and are returned + * unchanged. + */ +@Internal +public final class ColumnValueCodec { + + public static final String LZ4 = "lz4"; + public static final String ZSTD = "zstd"; + + public static final int HEADER_SIZE = 10; + public static final byte VERSION = 1; + public static final byte RAW_ENCODING = 0; + public static final byte LZ4_ENCODING = 1; + public static final byte ZSTD_ENCODING = 2; + + private static final int MAGIC = 0x46434D50; + private static final int MAGIC_SIZE = 4; + private static final int MAX_DECOMPRESSED_SIZE = 64 * 1024 * 1024; + + private final Map writeEncodings; + + /** Creates a codec for decoding self-describing column values. */ + public ColumnValueCodec() { + this(Collections.emptyMap()); + } + + /** Creates a codec using the compression configured in the table properties. */ + public ColumnValueCodec(Map tableConfigs) { + checkNotNull(tableConfigs, "tableConfigs must not be null"); + Map encodings = new HashMap<>(); + for (Map.Entry entry : tableConfigs.entrySet()) { + encodings.put(entry.getKey(), encodingOf(entry.getValue())); + } + this.writeEncodings = Collections.unmodifiableMap(encodings); + } + + /** Returns whether a property value enables the first supported codec. */ + public static boolean isLz4(String value) { + return value != null && LZ4.equals(value.trim().toLowerCase(Locale.ROOT)); + } + + /** Returns whether a property value enables ZSTD compression. */ + public static boolean isZstd(String value) { + return value != null && ZSTD.equals(value.trim().toLowerCase(Locale.ROOT)); + } + + /** Returns whether a property value is a supported compression. */ + public static boolean isSupportedCompression(String value) { + return isLz4(value) || isZstd(value); + } + + /** Returns the configured compression keyed by top-level column index. */ + public static Map compressionByColumn( + Schema schema, Map properties) { + checkNotNull(schema, "schema must not be null"); + checkNotNull(properties, "properties must not be null"); + Map compressionByColumn = new HashMap<>(); + for (int i = 0; i < schema.getColumns().size(); i++) { + String codec = properties.get(getColumnCompressionConfigKey(schema.getColumnName(i))); + if (codec != null) { + checkArgument( + isSupportedCompression(codec), + "Unsupported column compression '%s' for column '%s'.", + codec, + schema.getColumnName(i)); + checkArgument( + schema.getRowType().getTypeAt(i).getTypeRoot() == DataTypeRoot.BYTES, + "Column compression only supports BYTES columns, but column '%s' is %s.", + schema.getColumnName(i), + schema.getRowType().getTypeAt(i)); + compressionByColumn.put(i, codec); + } + } + return compressionByColumn; + } + + /** Returns the top-level column indexes configured for compression. */ + public static BitSet compressedColumnIndexes(Schema schema, Map properties) { + BitSet compressedColumns = new BitSet(schema.getColumns().size()); + for (Integer columnIndex : compressionByColumn(schema, properties).keySet()) { + compressedColumns.set(columnIndex); + } + return compressedColumns; + } + + /** Returns all top-level BYTES column indexes that may contain an encoded value. */ + public static BitSet bytesColumnIndexes(Schema schema) { + checkNotNull(schema, "schema must not be null"); + BitSet bytesColumns = new BitSet(schema.getColumns().size()); + for (int i = 0; i < schema.getColumns().size(); i++) { + if (schema.getRowType().getTypeAt(i).getTypeRoot() == DataTypeRoot.BYTES) { + bytesColumns.set(i); + } + } + return bytesColumns; + } + + /** Encodes a logical value using the configured compression and a self-describing envelope. */ + public byte[] encode(int columnIndex, byte[] value) { + checkNotNull(value, "value must not be null"); + Byte encoding = writeEncodings.get(columnIndex); + checkArgument( + encoding != null, "No compression configured for column index %s.", columnIndex); + byte[] compressed = compress(encoding, value); + if (compressed.length >= value.length) { + return envelope(RAW_ENCODING, value.length, value); + } + return envelope(encoding, value.length, compressed); + } + + /** + * Decodes a self-describing column value. Values without the magic are returned unchanged. + * + * @param value the physical value + * @return the logical value + * @throws CorruptRecordException if an envelope is malformed or cannot be decoded + */ + public byte[] decode(byte[] value) { + checkNotNull(value, "value must not be null"); + if (!hasMagic(value)) { + return value; + } + if (value.length < HEADER_SIZE) { + throw corrupt("Truncated column value envelope"); + } + if (value[4] != VERSION) { + throw corrupt("Unsupported column value envelope version: " + value[4]); + } + + int encoding = value[5] & 0xFF; + long decodedLength = Integer.toUnsignedLong(readInt(value, 6)); + if (decodedLength > MAX_DECOMPRESSED_SIZE) { + throw corrupt("Decoded column value size exceeds the limit: " + decodedLength); + } + + int outputLength = (int) decodedLength; + int payloadLength = value.length - HEADER_SIZE; + if (encoding == RAW_ENCODING) { + if (payloadLength != outputLength) { + throw corrupt( + "RAW column value length does not match the envelope: expected " + + outputLength + + ", actual " + + payloadLength); + } + return Arrays.copyOfRange(value, HEADER_SIZE, value.length); + } + + if (encoding != LZ4_ENCODING && encoding != ZSTD_ENCODING) { + throw corrupt("Unsupported column value encoding: " + encoding); + } + return decompress((byte) encoding, value, HEADER_SIZE, payloadLength, outputLength); + } + + /** Returns whether a value starts with the column-value envelope magic. */ + public boolean hasEnvelope(byte[] value) { + checkNotNull(value, "value must not be null"); + return hasMagic(value); + } + + private static byte[] envelope(byte encoding, int decodedLength, byte[] payload) { + byte[] envelope = new byte[HEADER_SIZE + payload.length]; + writeInt(envelope, 0, MAGIC); + envelope[4] = VERSION; + envelope[5] = encoding; + writeInt(envelope, 6, decodedLength); + System.arraycopy(payload, 0, envelope, HEADER_SIZE, payload.length); + return envelope; + } + + private static boolean hasMagic(byte[] value) { + return value.length >= MAGIC_SIZE && readInt(value, 0) == MAGIC; + } + + private static int readInt(byte[] value, int offset) { + return ((value[offset] & 0xFF) << 24) + | ((value[offset + 1] & 0xFF) << 16) + | ((value[offset + 2] & 0xFF) << 8) + | (value[offset + 3] & 0xFF); + } + + private static void writeInt(byte[] value, int offset, int number) { + value[offset] = (byte) (number >>> 24); + value[offset + 1] = (byte) (number >>> 16); + value[offset + 2] = (byte) (number >>> 8); + value[offset + 3] = (byte) number; + } + + private static CorruptRecordException corrupt(String message) { + return new CorruptRecordException(message); + } + + private static CorruptRecordException corrupt(String message, Throwable cause) { + return new CorruptRecordException(message, cause); + } + + private static byte encodingOf(String compression) { + checkArgument( + isSupportedCompression(compression), + "Unsupported column compression '%s'.", + compression); + return isLz4(compression) ? LZ4_ENCODING : ZSTD_ENCODING; + } + + private byte[] compress(byte encoding, byte[] value) { + ByteArrayOutputStream output = new ByteArrayOutputStream(); + try (InputStream input = new ByteArrayInputStream(value); + OutputStream compressedOutput = compressionOutputStream(output, encoding)) { + IOUtils.copyBytes(input, compressedOutput); + } catch (IOException e) { + throw new FlussRuntimeException( + "Failed to encode " + compressionName(encoding) + " column value.", e); + } + return output.toByteArray(); + } + + private byte[] decompress( + byte encoding, byte[] value, int offset, int length, int outputLength) { + ByteArrayOutputStream output = new ByteArrayOutputStream(outputLength); + try (InputStream input = compressionInputStream(encoding, value, offset, length)) { + IOUtils.copyBytes(input, output); + } catch (IOException | RuntimeException e) { + throw corrupt("Failed to decode " + compressionName(encoding) + " column value.", e); + } + byte[] result = output.toByteArray(); + if (result.length != outputLength) { + throw corrupt( + "Decoded column value length does not match the envelope: expected " + + outputLength + + ", actual " + + result.length); + } + return result; + } + + private static OutputStream compressionOutputStream(OutputStream output, byte encoding) + throws IOException { + if (encoding == LZ4_ENCODING) { + return new FlussLZ4BlockOutputStream(output); + } + if (encoding == ZSTD_ENCODING) { + return new FlussZSTDBlockOutputStream(output); + } + throw new IllegalArgumentException("Unsupported column value encoding: " + encoding); + } + + private static InputStream compressionInputStream( + byte encoding, byte[] value, int offset, int length) throws IOException { + if (encoding == LZ4_ENCODING) { + return new FlussLZ4BlockInputStream(ByteBuffer.wrap(value, offset, length)); + } + if (encoding == ZSTD_ENCODING) { + return new ZstdInputStream(new ByteArrayInputStream(value, offset, length)); + } + throw new IllegalArgumentException("Unsupported column value encoding: " + encoding); + } + + private static String compressionName(byte encoding) { + return encoding == LZ4_ENCODING ? "LZ4" : "ZSTD"; + } +} diff --git a/fluss-common/src/main/java/org/apache/fluss/compression/FlussZSTDBlockOutputStream.java b/fluss-common/src/main/java/org/apache/fluss/compression/FlussZSTDBlockOutputStream.java new file mode 100644 index 0000000000..abb26fb6d1 --- /dev/null +++ b/fluss-common/src/main/java/org/apache/fluss/compression/FlussZSTDBlockOutputStream.java @@ -0,0 +1,35 @@ +/* + * 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.fluss.compression; + +import com.github.luben.zstd.ZstdOutputStream; + +import java.io.IOException; +import java.io.OutputStream; + +/** An output stream that writes ZSTD frames using Fluss's default compression level. */ +public class FlussZSTDBlockOutputStream extends ZstdOutputStream { + + // TODO: Make the compression level configurable. + public static final int DEFAULT_COMPRESSION_LEVEL = 3; + + /** Creates a ZSTD output stream using Fluss's default compression level. */ + public FlussZSTDBlockOutputStream(OutputStream out) throws IOException { + super(out, DEFAULT_COMPRESSION_LEVEL); + } +} diff --git a/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java b/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java index 9a7ff3cd50..58bdcb5be9 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java @@ -30,10 +30,16 @@ import java.util.Map; import java.util.Optional; +import static org.apache.fluss.utils.Preconditions.checkArgument; +import static org.apache.fluss.utils.Preconditions.checkNotNull; + /** Utilities of Fluss {@link ConfigOptions}. */ @Internal public class FlussConfigUtils { + private static final String COLUMN_COMPRESSION_PREFIX = "fields."; + private static final String COLUMN_COMPRESSION_SUFFIX = ".compression"; + public static final Map> TABLE_OPTIONS; public static final Map> CLIENT_OPTIONS; public static final String TABLE_PREFIX = "table."; @@ -62,11 +68,35 @@ public class FlussConfigUtils { } public static boolean isTableStorageConfig(String key) { - return key.startsWith(TABLE_PREFIX); + return key.startsWith(TABLE_PREFIX) || isColumnCompressionConfig(key); } public static boolean isAlterableTableOption(String key) { - return ALTERABLE_TABLE_OPTIONS.contains(key); + return ALTERABLE_TABLE_OPTIONS.contains(key) || isColumnCompressionConfig(key); + } + + /** Returns the configuration key for a column compression setting. */ + public static String getColumnCompressionConfigKey(String columnName) { + checkNotNull(columnName, "columnName must not be null"); + checkArgument(!columnName.isEmpty(), "columnName must not be empty"); + return COLUMN_COMPRESSION_PREFIX + columnName + COLUMN_COMPRESSION_SUFFIX; + } + + /** Returns whether the key is a column compression configuration. */ + public static boolean isColumnCompressionConfig(String key) { + return key != null + && key.startsWith(COLUMN_COMPRESSION_PREFIX) + && key.endsWith(COLUMN_COMPRESSION_SUFFIX) + && key.length() + > COLUMN_COMPRESSION_PREFIX.length() + COLUMN_COMPRESSION_SUFFIX.length(); + } + + /** Returns the column name from a column compression configuration key. */ + public static String getColumnNameFromCompressionConfig(String key) { + checkArgument(isColumnCompressionConfig(key), "Not a column compression config: %s", key); + return key.substring( + COLUMN_COMPRESSION_PREFIX.length(), + key.length() - COLUMN_COMPRESSION_SUFFIX.length()); } /** diff --git a/fluss-common/src/main/java/org/apache/fluss/record/ValueRecordReadContext.java b/fluss-common/src/main/java/org/apache/fluss/record/ValueRecordReadContext.java index 368989ab0b..34608bf05d 100644 --- a/fluss-common/src/main/java/org/apache/fluss/record/ValueRecordReadContext.java +++ b/fluss-common/src/main/java/org/apache/fluss/record/ValueRecordReadContext.java @@ -21,7 +21,6 @@ import org.apache.fluss.metadata.Schema; import org.apache.fluss.metadata.SchemaGetter; import org.apache.fluss.row.decode.RowDecoder; -import org.apache.fluss.types.DataType; import java.util.HashMap; import java.util.Map; @@ -50,8 +49,7 @@ public RowDecoder getRowDecoder(int schemaId) { schemaId, (id) -> { Schema schema = schemaGetter.getSchema(schemaId); - return RowDecoder.create( - kvFormat, schema.getRowType().getChildren().toArray(new DataType[0])); + return RowDecoder.create(kvFormat, schema); }); } } diff --git a/fluss-common/src/main/java/org/apache/fluss/row/ColumnValueDecodingRow.java b/fluss-common/src/main/java/org/apache/fluss/row/ColumnValueDecodingRow.java new file mode 100644 index 0000000000..3fdcf719d2 --- /dev/null +++ b/fluss-common/src/main/java/org/apache/fluss/row/ColumnValueDecodingRow.java @@ -0,0 +1,148 @@ +/* + * 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.fluss.row; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.compression.ColumnValueCodec; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** An {@link InternalRow} that lazily decodes self-describing BYTES column values. */ +@Internal +public final class ColumnValueDecodingRow implements InternalRow { + + private static final ColumnValueCodec COLUMN_VALUE_CODEC = new ColumnValueCodec(); + + private final InternalRow row; + + private ColumnValueDecodingRow(InternalRow row) { + this.row = checkNotNull(row, "row must not be null"); + } + + /** Wraps a row unless it is already lazily decoding column values. */ + public static InternalRow wrap(InternalRow row) { + if (row instanceof ColumnValueDecodingRow) { + return row; + } + return new ColumnValueDecodingRow(row); + } + + @Override + public int getFieldCount() { + return row.getFieldCount(); + } + + @Override + public boolean isNullAt(int pos) { + return row.isNullAt(pos); + } + + @Override + public boolean getBoolean(int pos) { + return row.getBoolean(pos); + } + + @Override + public byte getByte(int pos) { + return row.getByte(pos); + } + + @Override + public short getShort(int pos) { + return row.getShort(pos); + } + + @Override + public int getInt(int pos) { + return row.getInt(pos); + } + + @Override + public long getLong(int pos) { + return row.getLong(pos); + } + + @Override + public float getFloat(int pos) { + return row.getFloat(pos); + } + + @Override + public double getDouble(int pos) { + return row.getDouble(pos); + } + + @Override + public BinaryString getChar(int pos, int length) { + return row.getChar(pos, length); + } + + @Override + public BinaryString getString(int pos) { + return row.getString(pos); + } + + @Override + public Decimal getDecimal(int pos, int precision, int scale) { + return row.getDecimal(pos, precision, scale); + } + + @Override + public TimestampNtz getTimestampNtz(int pos, int precision) { + return row.getTimestampNtz(pos, precision); + } + + @Override + public TimestampLtz getTimestampLtz(int pos, int precision) { + return row.getTimestampLtz(pos, precision); + } + + @Override + public byte[] getBinary(int pos, int length) { + return row.getBinary(pos, length); + } + + @Override + public byte[] getBytes(int pos) { + if (row.isNullAt(pos)) { + return null; + } + return COLUMN_VALUE_CODEC.decode(row.getBytes(pos)); + } + + @Override + public InternalArray getArray(int pos) { + return row.getArray(pos); + } + + @Override + public InternalMap getMap(int pos) { + return row.getMap(pos); + } + + @Override + public InternalRow getRow(int pos, int numFields) { + return row.getRow(pos, numFields); + } + + @Override + public String toString() { + return row.toString(); + } +} diff --git a/fluss-common/src/main/java/org/apache/fluss/row/decode/AbstractRowDecoder.java b/fluss-common/src/main/java/org/apache/fluss/row/decode/AbstractRowDecoder.java new file mode 100644 index 0000000000..d17ac1aa81 --- /dev/null +++ b/fluss-common/src/main/java/org/apache/fluss/row/decode/AbstractRowDecoder.java @@ -0,0 +1,95 @@ +/* + * 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.fluss.row.decode; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.compression.ColumnValueCodec; +import org.apache.fluss.metadata.KvFormat; +import org.apache.fluss.row.BinaryRow; +import org.apache.fluss.row.InternalRow; +import org.apache.fluss.row.encode.RowEncoder; +import org.apache.fluss.types.DataType; + +import java.util.BitSet; +import java.util.Collections; +import java.util.Map; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** Base row decoder that restores encoded column values after decoding the physical row. */ +@Internal +public abstract class AbstractRowDecoder implements RowDecoder { + + private final KvFormat kvFormat; + private final DataType[] fieldDataTypes; + private final BitSet decodedColumns; + private final ColumnValueCodec columnValueCodec; + private final InternalRow.FieldGetter[] fieldGetters; + + protected AbstractRowDecoder(KvFormat kvFormat, DataType[] fieldDataTypes) { + this(kvFormat, fieldDataTypes, new BitSet(), Collections.emptyMap()); + } + + protected AbstractRowDecoder( + KvFormat kvFormat, + DataType[] fieldDataTypes, + BitSet decodedColumns, + Map tableConfigs) { + this.kvFormat = checkNotNull(kvFormat, "kvFormat must not be null"); + this.fieldDataTypes = checkNotNull(fieldDataTypes, "fieldDataTypes must not be null"); + this.decodedColumns = + (BitSet) checkNotNull(decodedColumns, "decodedColumns must not be null").clone(); + this.columnValueCodec = new ColumnValueCodec(tableConfigs); + this.fieldGetters = new InternalRow.FieldGetter[fieldDataTypes.length]; + for (int i = 0; i < fieldDataTypes.length; i++) { + this.fieldGetters[i] = InternalRow.createFieldGetter(fieldDataTypes[i], i); + } + } + + /** Restores encoded fields and returns a row in the same physical format. */ + protected final BinaryRow decodeFields(BinaryRow row) { + if (decodedColumns.isEmpty()) { + return row; + } + + Object[] values = new Object[fieldDataTypes.length]; + boolean changed = false; + for (int i = 0; i < fieldDataTypes.length; i++) { + Object value = fieldGetters[i].getFieldOrNull(row); + if (value != null && decodedColumns.get(i)) { + byte[] bytes = (byte[]) value; + if (columnValueCodec.hasEnvelope(bytes)) { + value = columnValueCodec.decode(bytes); + changed = true; + } + } + values[i] = value; + } + if (!changed) { + return row; + } + + RowEncoder encoder = RowEncoder.create(kvFormat, fieldDataTypes); + encoder.startNewRow(); + for (int i = 0; i < values.length; i++) { + encoder.encodeField(i, values[i]); + } + return encoder.finishRow(); + } +} diff --git a/fluss-common/src/main/java/org/apache/fluss/row/decode/CompactedRowDecoder.java b/fluss-common/src/main/java/org/apache/fluss/row/decode/CompactedRowDecoder.java index e66f590958..b372199995 100644 --- a/fluss-common/src/main/java/org/apache/fluss/row/decode/CompactedRowDecoder.java +++ b/fluss-common/src/main/java/org/apache/fluss/row/decode/CompactedRowDecoder.java @@ -18,30 +18,41 @@ package org.apache.fluss.row.decode; import org.apache.fluss.memory.MemorySegment; +import org.apache.fluss.metadata.KvFormat; import org.apache.fluss.row.compacted.CompactedRow; import org.apache.fluss.row.compacted.CompactedRowDeserializer; import org.apache.fluss.types.DataType; +import java.util.BitSet; +import java.util.Collections; +import java.util.Map; + /** A decoder to decode {@link CompactedRow} from a byte array or memory segment. */ -public class CompactedRowDecoder implements RowDecoder { +public class CompactedRowDecoder extends AbstractRowDecoder { private final DataType[] fieldDataTypes; private final CompactedRowDeserializer deserializer; public CompactedRowDecoder(DataType[] fieldDataTypes) { + this(fieldDataTypes, new BitSet(), Collections.emptyMap()); + } + + CompactedRowDecoder( + DataType[] fieldDataTypes, BitSet decodedColumns, Map tableConfigs) { + super(KvFormat.COMPACTED, fieldDataTypes, decodedColumns, tableConfigs); this.fieldDataTypes = fieldDataTypes; this.deserializer = new CompactedRowDeserializer(fieldDataTypes); } @Override public CompactedRow decode(byte[] values) { - return CompactedRow.from(fieldDataTypes, values, deserializer); + return (CompactedRow) decodeFields(CompactedRow.from(fieldDataTypes, values, deserializer)); } @Override public CompactedRow decode(MemorySegment segment, int offset, int sizeInBytes) { CompactedRow compactedRow = new CompactedRow(fieldDataTypes.length, deserializer); compactedRow.pointTo(segment, offset, sizeInBytes); - return compactedRow; + return (CompactedRow) decodeFields(compactedRow); } } diff --git a/fluss-common/src/main/java/org/apache/fluss/row/decode/FixedSchemaDecoder.java b/fluss-common/src/main/java/org/apache/fluss/row/decode/FixedSchemaDecoder.java index faa42df1fd..f95c45b612 100644 --- a/fluss-common/src/main/java/org/apache/fluss/row/decode/FixedSchemaDecoder.java +++ b/fluss-common/src/main/java/org/apache/fluss/row/decode/FixedSchemaDecoder.java @@ -24,7 +24,6 @@ import org.apache.fluss.row.InternalRow; import org.apache.fluss.row.ProjectedRow; import org.apache.fluss.row.encode.KvValueLayout; -import org.apache.fluss.types.DataType; import org.apache.fluss.utils.SchemaUtil; /** @@ -48,9 +47,7 @@ public class FixedSchemaDecoder { private final boolean noProjection; public FixedSchemaDecoder(KvFormat kvFormat, Schema sourceSchema, Schema targetSchema) { - this.rowDecoder = - RowDecoder.create( - kvFormat, sourceSchema.getRowType().getChildren().toArray(new DataType[0])); + this.rowDecoder = RowDecoder.create(kvFormat, sourceSchema); this.fieldIdMapping = SchemaUtil.getIndexMapping(sourceSchema, targetSchema); this.noProjection = false; } @@ -60,9 +57,7 @@ public FixedSchemaDecoder(KvFormat kvFormat, Schema sourceSchema, Schema targetS * target schema. */ public FixedSchemaDecoder(KvFormat kvFormat, Schema schema) { - this.rowDecoder = - RowDecoder.create( - kvFormat, schema.getRowType().getChildren().toArray(new DataType[0])); + this.rowDecoder = RowDecoder.create(kvFormat, schema); this.fieldIdMapping = null; this.noProjection = true; } diff --git a/fluss-common/src/main/java/org/apache/fluss/row/decode/IndexedRowDecoder.java b/fluss-common/src/main/java/org/apache/fluss/row/decode/IndexedRowDecoder.java index b4f007f80a..36ee559cce 100644 --- a/fluss-common/src/main/java/org/apache/fluss/row/decode/IndexedRowDecoder.java +++ b/fluss-common/src/main/java/org/apache/fluss/row/decode/IndexedRowDecoder.java @@ -18,26 +18,37 @@ package org.apache.fluss.row.decode; import org.apache.fluss.memory.MemorySegment; +import org.apache.fluss.metadata.KvFormat; import org.apache.fluss.row.indexed.IndexedRow; import org.apache.fluss.types.DataType; +import java.util.BitSet; +import java.util.Collections; +import java.util.Map; + /** A decoder to decode {@link IndexedRow} from a byte array or memory segment. */ -public class IndexedRowDecoder implements RowDecoder { +public class IndexedRowDecoder extends AbstractRowDecoder { private final DataType[] fieldDataTypes; public IndexedRowDecoder(DataType[] fieldDataTypes) { + this(fieldDataTypes, new BitSet(), Collections.emptyMap()); + } + + IndexedRowDecoder( + DataType[] fieldDataTypes, BitSet decodedColumns, Map tableConfigs) { + super(KvFormat.INDEXED, fieldDataTypes, decodedColumns, tableConfigs); this.fieldDataTypes = fieldDataTypes; } @Override public IndexedRow decode(byte[] values) { - return IndexedRow.from(fieldDataTypes, values); + return (IndexedRow) decodeFields(IndexedRow.from(fieldDataTypes, values)); } @Override public IndexedRow decode(MemorySegment segment, int offset, int sizeInBytes) { IndexedRow indexedRow = new IndexedRow(fieldDataTypes); indexedRow.pointTo(segment, offset, sizeInBytes); - return indexedRow; + return (IndexedRow) decodeFields(indexedRow); } } diff --git a/fluss-common/src/main/java/org/apache/fluss/row/decode/RowDecoder.java b/fluss-common/src/main/java/org/apache/fluss/row/decode/RowDecoder.java index e257d1998b..c158e20100 100644 --- a/fluss-common/src/main/java/org/apache/fluss/row/decode/RowDecoder.java +++ b/fluss-common/src/main/java/org/apache/fluss/row/decode/RowDecoder.java @@ -18,13 +18,18 @@ package org.apache.fluss.row.decode; import org.apache.fluss.annotation.PublicEvolving; +import org.apache.fluss.compression.ColumnValueCodec; import org.apache.fluss.memory.MemorySegment; import org.apache.fluss.metadata.KvFormat; +import org.apache.fluss.metadata.Schema; import org.apache.fluss.record.ValueRecord; import org.apache.fluss.row.BinaryRow; import org.apache.fluss.row.InternalRow; import org.apache.fluss.types.DataType; +import java.util.BitSet; +import java.util.Collections; + /** * A decoder to read {@link BinaryRow binary format InternalRow} from a byte array or memory segment * of a value record {@link ValueRecord}. @@ -45,6 +50,19 @@ static RowDecoder create(KvFormat kvFormat, DataType[] fieldDataTypes) { } } + /** Creates a row decoder that restores self-describing column values. */ + static RowDecoder create(KvFormat kvFormat, Schema schema) { + DataType[] fieldDataTypes = schema.getRowType().getChildren().toArray(new DataType[0]); + BitSet decodedColumns = ColumnValueCodec.bytesColumnIndexes(schema); + if (kvFormat == KvFormat.COMPACTED) { + return new CompactedRowDecoder(fieldDataTypes, decodedColumns, Collections.emptyMap()); + } else if (kvFormat == KvFormat.INDEXED) { + return new IndexedRowDecoder(fieldDataTypes, decodedColumns, Collections.emptyMap()); + } else { + throw new IllegalArgumentException("Unsupported kv format: " + kvFormat); + } + } + /** Decode the byte array to {@link BinaryRow}. */ BinaryRow decode(byte[] values); diff --git a/fluss-common/src/main/java/org/apache/fluss/row/encode/AbstractRowEncoder.java b/fluss-common/src/main/java/org/apache/fluss/row/encode/AbstractRowEncoder.java new file mode 100644 index 0000000000..d69f9e8bdb --- /dev/null +++ b/fluss-common/src/main/java/org/apache/fluss/row/encode/AbstractRowEncoder.java @@ -0,0 +1,58 @@ +/* + * 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.fluss.row.encode; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.compression.ColumnValueCodec; + +import java.util.BitSet; +import java.util.Map; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** Base row encoder that applies column-value encoding before writing the physical row. */ +@Internal +public abstract class AbstractRowEncoder implements RowEncoder { + + private final BitSet encodedColumns; + private final ColumnValueCodec columnValueCodec; + + protected AbstractRowEncoder(BitSet encodedColumns, Map tableConfigs) { + this.encodedColumns = + (BitSet) checkNotNull(encodedColumns, "encodedColumns must not be null").clone(); + this.columnValueCodec = new ColumnValueCodec(tableConfigs); + } + + @Override + public final boolean hasColumnValueEncoding() { + return !encodedColumns.isEmpty(); + } + + @Override + public final void encodeField(int pos, Object value) { + Object encodedValue = value; + if (value != null && encodedColumns.get(pos)) { + encodedValue = columnValueCodec.encode(pos, (byte[]) value); + } + encode(pos, encodedValue); + } + + /** Writes a field to the concrete physical row format. */ + protected abstract void encode(int pos, Object value); +} diff --git a/fluss-common/src/main/java/org/apache/fluss/row/encode/AlignedRowEncoder.java b/fluss-common/src/main/java/org/apache/fluss/row/encode/AlignedRowEncoder.java index 0bee6f4c36..27e19240e5 100644 --- a/fluss-common/src/main/java/org/apache/fluss/row/encode/AlignedRowEncoder.java +++ b/fluss-common/src/main/java/org/apache/fluss/row/encode/AlignedRowEncoder.java @@ -25,6 +25,9 @@ import org.apache.fluss.row.aligned.AlignedRowWriter; import org.apache.fluss.types.DataType; +import java.util.BitSet; +import java.util.Collections; + import static org.apache.fluss.row.BinaryRow.BinaryRowFormat.ALIGNED; /** @@ -32,12 +35,13 @@ * * @since 0.9 */ -public class AlignedRowEncoder implements RowEncoder { +public class AlignedRowEncoder extends AbstractRowEncoder { private final AlignedRow reuseRow; private final AlignedRowWriter reuseWriter; private final BinaryWriter.ValueWriter[] valueWriters; public AlignedRowEncoder(DataType[] fieldTypes) { + super(new BitSet(), Collections.emptyMap()); this.reuseRow = new AlignedRow(fieldTypes.length); this.reuseWriter = new AlignedRowWriter(reuseRow); this.valueWriters = new BinaryWriter.ValueWriter[fieldTypes.length]; @@ -52,7 +56,7 @@ public void startNewRow() { } @Override - public void encodeField(int pos, Object value) { + protected void encode(int pos, Object value) { valueWriters[pos].writeValue(reuseWriter, pos, value); } diff --git a/fluss-common/src/main/java/org/apache/fluss/row/encode/CompactedRowEncoder.java b/fluss-common/src/main/java/org/apache/fluss/row/encode/CompactedRowEncoder.java index 546ce45a28..e51aef34ce 100644 --- a/fluss-common/src/main/java/org/apache/fluss/row/encode/CompactedRowEncoder.java +++ b/fluss-common/src/main/java/org/apache/fluss/row/encode/CompactedRowEncoder.java @@ -24,6 +24,10 @@ import org.apache.fluss.row.compacted.CompactedRowWriter; import org.apache.fluss.types.DataType; +import java.util.BitSet; +import java.util.Collections; +import java.util.Map; + import static org.apache.fluss.row.BinaryRow.BinaryRowFormat.COMPACTED; /** @@ -32,7 +36,7 @@ * @since 0.2 */ @PublicEvolving -public class CompactedRowEncoder implements RowEncoder { +public class CompactedRowEncoder extends AbstractRowEncoder { private final DataType[] fieldDataTypes; private final CompactedRowWriter writer; @@ -40,6 +44,12 @@ public class CompactedRowEncoder implements RowEncoder { private final CompactedRowDeserializer compactedRowDeserializer; public CompactedRowEncoder(DataType[] fieldDataTypes) { + this(fieldDataTypes, new BitSet(), Collections.emptyMap()); + } + + CompactedRowEncoder( + DataType[] fieldDataTypes, BitSet encodedColumns, Map tableConfigs) { + super(encodedColumns, tableConfigs); this.fieldDataTypes = fieldDataTypes; // writer for row's fields writer = new CompactedRowWriter(fieldDataTypes.length); @@ -56,7 +66,7 @@ public void startNewRow() { } @Override - public void encodeField(int pos, Object value) { + protected void encode(int pos, Object value) { fieldWriters[pos].writeValue(writer, pos, value); } diff --git a/fluss-common/src/main/java/org/apache/fluss/row/encode/IndexedRowEncoder.java b/fluss-common/src/main/java/org/apache/fluss/row/encode/IndexedRowEncoder.java index 53fc9fac14..44ec8db79f 100644 --- a/fluss-common/src/main/java/org/apache/fluss/row/encode/IndexedRowEncoder.java +++ b/fluss-common/src/main/java/org/apache/fluss/row/encode/IndexedRowEncoder.java @@ -24,6 +24,10 @@ import org.apache.fluss.types.DataType; import org.apache.fluss.types.RowType; +import java.util.BitSet; +import java.util.Collections; +import java.util.Map; + import static org.apache.fluss.row.BinaryRow.BinaryRowFormat.INDEXED; /** @@ -32,7 +36,7 @@ * @since 0.2 */ @PublicEvolving -public class IndexedRowEncoder implements RowEncoder { +public class IndexedRowEncoder extends AbstractRowEncoder { private final DataType[] fieldDataTypes; private final IndexedRowWriter rowWriter; @@ -43,6 +47,12 @@ public IndexedRowEncoder(RowType rowType) { } public IndexedRowEncoder(DataType[] fieldDataTypes) { + this(fieldDataTypes, new BitSet(), Collections.emptyMap()); + } + + IndexedRowEncoder( + DataType[] fieldDataTypes, BitSet encodedColumns, Map tableConfigs) { + super(encodedColumns, tableConfigs); this.fieldDataTypes = fieldDataTypes; // create writer. this.fieldWriters = new BinaryWriter.ValueWriter[fieldDataTypes.length]; @@ -58,7 +68,7 @@ public void startNewRow() { } @Override - public void encodeField(int pos, Object value) { + protected void encode(int pos, Object value) { fieldWriters[pos].writeValue(rowWriter, pos, value); } diff --git a/fluss-common/src/main/java/org/apache/fluss/row/encode/RowEncoder.java b/fluss-common/src/main/java/org/apache/fluss/row/encode/RowEncoder.java index f18afd3607..e73133e217 100644 --- a/fluss-common/src/main/java/org/apache/fluss/row/encode/RowEncoder.java +++ b/fluss-common/src/main/java/org/apache/fluss/row/encode/RowEncoder.java @@ -18,13 +18,18 @@ package org.apache.fluss.row.encode; import org.apache.fluss.annotation.PublicEvolving; +import org.apache.fluss.compression.ColumnValueCodec; import org.apache.fluss.memory.MemorySegment; import org.apache.fluss.metadata.KvFormat; +import org.apache.fluss.metadata.Schema; import org.apache.fluss.row.BinaryRow; import org.apache.fluss.row.InternalRow; import org.apache.fluss.types.DataType; import org.apache.fluss.types.RowType; +import java.util.BitSet; +import java.util.Map; + /** * An encoder to write {@link BinaryRow binary format InternalRow}. It's used to write row * multi-times one by one. When writing a new row: @@ -62,6 +67,30 @@ static RowEncoder create(KvFormat kvFormat, DataType[] fieldDataTypes) { } } + /** Creates a row encoder that applies the table's configured column-value encoding. */ + static RowEncoder create( + KvFormat kvFormat, Schema schema, Map tableProperties) { + DataType[] fieldDataTypes = schema.getRowType().getChildren().toArray(new DataType[0]); + Map compressionByColumn = + ColumnValueCodec.compressionByColumn(schema, tableProperties); + BitSet encodedColumns = new BitSet(schema.getColumns().size()); + for (Integer columnIndex : compressionByColumn.keySet()) { + encodedColumns.set(columnIndex); + } + if (kvFormat == KvFormat.COMPACTED) { + return new CompactedRowEncoder(fieldDataTypes, encodedColumns, compressionByColumn); + } else if (kvFormat == KvFormat.INDEXED) { + return new IndexedRowEncoder(fieldDataTypes, encodedColumns, compressionByColumn); + } else { + throw new IllegalArgumentException("Unsupported kv format: " + kvFormat); + } + } + + /** Returns whether this encoder transforms configured column values. */ + default boolean hasColumnValueEncoding() { + return false; + } + /** Start to write a new row. */ void startNewRow(); diff --git a/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueDecoder.java b/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueDecoder.java index 2c0727a275..46a76daa99 100644 --- a/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueDecoder.java +++ b/fluss-common/src/main/java/org/apache/fluss/row/encode/ValueDecoder.java @@ -24,7 +24,6 @@ import org.apache.fluss.record.BinaryValue; import org.apache.fluss.row.BinaryRow; import org.apache.fluss.row.decode.RowDecoder; -import org.apache.fluss.types.DataType; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @@ -59,9 +58,7 @@ public BinaryValue decodeValue(byte[] valueBytes) { schemaId, (id) -> { Schema schema = schemaGetter.getSchema(schemaId); - return RowDecoder.create( - kvFormat, - schema.getRowType().getChildren().toArray(new DataType[0])); + return RowDecoder.create(kvFormat, schema); }); BinaryRow row = diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java index 8ba8204a54..c2f18d45e3 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java @@ -18,6 +18,7 @@ package org.apache.fluss.server.utils; import org.apache.fluss.annotation.Internal; +import org.apache.fluss.compression.ColumnValueCodec; import org.apache.fluss.config.AutoPartitionTimeUnit; import org.apache.fluss.config.ConfigOption; import org.apache.fluss.config.ConfigOptions; @@ -54,13 +55,16 @@ import java.util.EnumSet; import java.util.LinkedHashSet; import java.util.List; +import java.util.Map; import java.util.Optional; import java.util.Set; import java.util.stream.Collectors; import static org.apache.fluss.config.ConfigOptions.CURRENT_KV_FORMAT_VERSION; import static org.apache.fluss.config.FlussConfigUtils.TABLE_OPTIONS; +import static org.apache.fluss.config.FlussConfigUtils.getColumnNameFromCompressionConfig; import static org.apache.fluss.config.FlussConfigUtils.isAlterableTableOption; +import static org.apache.fluss.config.FlussConfigUtils.isColumnCompressionConfig; import static org.apache.fluss.config.FlussConfigUtils.isTableStorageConfig; import static org.apache.fluss.config.StatisticsConfigUtils.validateStatisticsConfig; import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME; @@ -107,6 +111,10 @@ public static void validateTableDescriptor( continue; } + if (isColumnCompressionConfig(key)) { + continue; + } + if (isTableStorageConfig(key)) { throw new InvalidConfigException( String.format( @@ -137,6 +145,7 @@ public static void validateTableDescriptor( checkKvValueLayout(tableConf, hasPrimaryKey); checkPartition(tableConf, tableDescriptor.getPartitionKeys(), schema.getRowType()); checkSystemColumns(schema.getRowType()); + checkColumnCompression(tableDescriptor, tableConf, hasPrimaryKey); validateStatisticsConfig(tableDescriptor); checkTableLakeFormatMatchesCluster(tableConf, clusterDataLakeFormat); } @@ -520,6 +529,54 @@ private static void checkArrowCompression(Configuration tableConf) { } } + private static void checkColumnCompression( + TableDescriptor tableDescriptor, Configuration tableConf, boolean hasPrimaryKey) { + for (Map.Entry property : tableConf.toMap().entrySet()) { + if (!isColumnCompressionConfig(property.getKey())) { + continue; + } + + String columnName = getColumnNameFromCompressionConfig(property.getKey()); + if (!ColumnValueCodec.isSupportedCompression(property.getValue())) { + throw new InvalidConfigException( + String.format( + "Unsupported compression '%s' for column '%s'. Supported compressions: lz4, zstd.", + property.getValue(), columnName)); + } + if (!hasPrimaryKey) { + throw new InvalidTableException( + "Column compression is only supported for primary key tables."); + } + + Schema schema = tableDescriptor.getSchema(); + int columnIndex = schema.getRowType().getFieldIndex(columnName); + if (columnIndex < 0) { + throw new InvalidConfigException( + String.format( + "Compression property '%s' refers to unknown column '%s'.", + property.getKey(), columnName)); + } + if (schema.getRowType().getTypeAt(columnIndex).getTypeRoot() != DataTypeRoot.BYTES) { + throw new InvalidConfigException( + String.format( + "Column compression only supports BYTES columns, but column '%s' is %s.", + columnName, schema.getRowType().getTypeAt(columnIndex))); + } + if (schema.getPrimaryKeyColumnNames().contains(columnName) + || tableDescriptor.getBucketKeys().contains(columnName) + || tableDescriptor.getPartitionKeys().contains(columnName)) { + throw new InvalidConfigException( + String.format( + "Column compression is not supported for key column '%s'.", + columnName)); + } + if (tableConf.get(ConfigOptions.TABLE_MERGE_ENGINE) == MergeEngineType.AGGREGATION) { + throw new InvalidConfigException( + "Column compression is not supported with the aggregation merge engine."); + } + } + } + private static void checkMergeEngine( Configuration tableConf, boolean hasPrimaryKey, Schema schema) { MergeEngineType mergeEngine = tableConf.get(ConfigOptions.TABLE_MERGE_ENGINE); diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkConversions.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkConversions.scala index eeb726678d..21053479df 100644 --- a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkConversions.scala +++ b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkConversions.scala @@ -78,7 +78,7 @@ object SparkConversions { val (tableProps, customProps) = caseInsensitiveProps.filterNot(e => SPARK_TABLE_OPTIONS.contains(e._1)).partition { - case (key, _) => key.startsWith(FlussConfigUtils.TABLE_PREFIX) + case (key, _) => FlussConfigUtils.isTableStorageConfig(key) } tableDescriptorBuilder