all tables in lake storage configured cluster should use lake's bucke… #362
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
luoyuxia
force-pushed
the
use-lake-bucket-assigner
branch
5 times, most recently
from
84ab0a0 to
b69bc6b
Compare
luoyuxia
force-pushed
the
use-lake-bucket-assigner
branch
from
b69bc6b to
12e2d38
Compare
wuchong
requested changes
Feb 13, 2025
There was a problem hiding this comment.
Some high-level design suggestions:
- The cluster datalake configuration should align with table configuration, how about:
datalake.format = paimon/iceberg
datalake.paimon.catalog = hive
datalake.paimon.metastore = thrift://localhost:9083
datalake.paimon.warehouse = hdfs://localhost:9000/user/hive/warehouse
- Besides, the
table.datalake.format, we can also add the catalog and metastore information into table config, so that the table can be self-contained, and connectors don't need to retrievedescribeLakeStorage(describeLakeStoragecan be removed then). If there is any auth/securtiy tokens to retrieve, we can add another RPC similar togetFileSystemSecurityToken, but that should be optional.
table.datalake.enabled = true/false
table.datalake.format = paimon/iceberg
// logical properties
table.datalake.paimon.catalog = hive
table.datalake.paimon.metastore = thrift://localhost:9083
table.datalake.paimon.warehouse = hdfs://localhost:9000/user/hive/warehouse
Considering the metastore location may be changed over time, it would be a logical property that is added during getTableInfo instead of a physical property stored in zookeeper. And thus metastore properties shouldn't be set or altered.
- It's confusing that the key encoding is different for bucketing and storing. It's meaningless to keep our own key encoding when it is a datalake-enabled table. We also discussed that bucket-rescale can be supported even if we use Paimon/Iceberg key encoding/bucketing. So we can introduce two separate interfaces for encoding and bucketing.
interface KeyEncoder {
byte[] encodeKey(InternalRow row);
}
interface BucketAssigner {
int assignBucket(@Nullable byte[] bucketKey, Cluster cluster);
}- We should add tests to verify the encoding and bucketing is the same with Paimon
fluss-client/src/main/java/com/alibaba/fluss/client/write/WriterClient.java
Outdated
Show resolved
Hide resolved
fluss-client/src/main/java/com/alibaba/fluss/client/table/writer/UpsertWriterImpl.java
Outdated
Show resolved
Hide resolved
fluss-common/src/main/java/com/alibaba/fluss/config/TableConfig.java
Outdated
Show resolved
Hide resolved
luoyuxia
force-pushed
the
use-lake-bucket-assigner
branch
7 times, most recently
from
021edb8 to
840980a
Compare
luoyuxia
force-pushed
the
use-lake-bucket-assigner
branch
from
840980a to
4a8108f
Compare
…ated in DataLake configured cluster (alibaba#362)
wuchong
reviewed
Feb 21, 2025
| @@ -124,4 +124,14 @@ public static Map<String, String> extractConfigStartWith( | |||
| } | |||
| return extractedConfig; | |||
| } | |||
|
|
|||
| private static Map<String, String> normalizeToPaimonConfigs( | |||
There was a problem hiding this comment.
Use datalake.paimon.metastore to replace datalake.paimon.catalog.
| datalakeFormat.toString(), normalizeToPaimonConfigs(datalakeConfig)); | ||
| } | ||
|
|
||
| private static Map<String, String> normalizeToPaimonConfigs( |
There was a problem hiding this comment.
Use datalake.paimon.metastore to replace datalake.paimon.catalog
| @@ -186,6 +194,14 @@ private TableDescriptor applySystemDefaults(TableDescriptor tableDescriptor) { | |||
| newDescriptor = newDescriptor.withReplicationFactor(defaultReplicationFactor); | |||
| } | |||
|
|
|||
| // if lake storage is not null, we need to add the datalake type | |||
| // to the property of the table | |||
| if (dataLakeFormat != null) { | |||
There was a problem hiding this comment.
we should throw exception if the table is datalake enabled, but cluster not.
| this.lakeTableBucketAssigner = null; | ||
| this.bucketKeyEncoder = | ||
| CompactedKeyEncoder.createKeyEncoder(lookupRowType, tableInfo.getBucketKeys()); | ||
| this.lakeBucketAssigner = null; |
There was a problem hiding this comment.
Constructing the bucketKeyEncoder is still very complex. However, we can unify the default and lake assigners into one util method.
| } else { | ||
| return lakeTableBucketAssigner.assignBucket( | ||
| keyBytes, key, metadataUpdater.getCluster()); | ||
| return lakeBucketAssigner.assignBucket(bucketKeyBytes); |
There was a problem hiding this comment.
It is still not unified and complex to assign bucket. We can introduce a BucketingFunction#bucketing(byte[] key, int numBuckets) to unify the bucket assigned instead of using a util method and passing optional lake bucket assigner. We should hold only one not-null bucket assigner, that will be easier to use.
wuchong
force-pushed
the
use-lake-bucket-assigner
branch
from
4a8108f to
fffb61c
Compare
|
Please review the changes I just pushed @luoyuxia . |
…lake enabled or not
wuchong
force-pushed
the
use-lake-bucket-assigner
branch
from
fffb61c to
e5f1628
Compare
wuchong
approved these changes
Feb 22, 2025
wuchong
pushed a commit
that referenced
this pull request
Feb 22, 2025
…ated in DataLake configured cluster (#362)
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
…t assigner
Purpose
Linked issue: close #343
Make all tables in lake storage configured cluster use lake's bucket assign strategy..
table.datalake.formatto the corresponding datalaketable.datalake.format, use lake's bucket assinger to assign bucket for rowsStaticBucketAssignerandDynamicBucketAssigner.. For tables with bucket keys,use
StaticBucketAssignerto caclcuate bucket id during convert row intoWriteRecord.Tests
LakeTableManagerITCase to verify
table.datalake.formatwill be set in lakestorage configured clusterFlussLakeTableITCase to verfy write/lookup will use lake's bucket assinger t
API and Format
Documentation