You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "XiaoHongbo-Hope (via GitHub)" <gi...@apache.org> on 2026/09/02 10:16:50 UTC
[PR] feat(go): add map blob readers [paimon-rust]
XiaoHongbo-Hope opened a new pull request, #784:
URL: https://github.com/apache/paimon-rust/pull/784
## Summary
- add generic `ReadBlobMap` and `ReadBlobMaps` helpers
- resolve every map value through one existing `BlobReader.ReadBlobs` call
- preserve keys and nil/empty maps
The Go layer only flattens and restores maps. Descriptor parsing, range merging, concurrency limits, FileIO, and DLF data-token handling remain in Rust core.
## Testing
- `go test -v -run "^TestBlob" ./...`
- `go vet ./...`
- read-only DLF E2E using a temporary BLOB table in `morax_test`: wrote two values, read their descriptors, resolved two maps (including a duplicate descriptor) through a table-backed Go `BlobReader`, and verified keys and bytes
The temporary DLF database was removed after the test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(blob): support MAP reads in Go [paimon-rust]
Posted by "QuakeWang (via GitHub)" <gi...@apache.org>.
QuakeWang commented on code in PR #784:
URL: https://github.com/apache/paimon-rust/pull/784#discussion_r3914753055
##########
docs/src/go-binding.md:
##########
@@ -147,8 +147,8 @@ catalog, err := paimon.NewCatalog(map[string]string{
"token.provider": "dlf",
"dlf.region": os.Getenv("DLF_REGION"),
"dlf.oss-endpoint": os.Getenv("DLF_OSS_ENDPOINT"),
- "dlf.token-loader": "ecs",
- "dlf.token-ecs-role-name": os.Getenv("DLF_ECS_ROLE"),
+ "dlf.access-key-id": os.Getenv("DLF_ACCESS_KEY_ID"),
Review Comment:
This changes the DLF example from the ECS RAM-role loader to static credentials, which is unrelated to MAP<K, BLOB> support and drops automatic credential refresh. Please revert it here or move it to a separate PR with its own rationale.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(blob): support MAP reads in Go [paimon-rust]
Posted by "QuakeWang (via GitHub)" <gi...@apache.org>.
QuakeWang commented on code in PR #784:
URL: https://github.com/apache/paimon-rust/pull/784#discussion_r3914749652
##########
crates/paimon/src/arrow/format/blob.rs:
##########
@@ -764,6 +995,277 @@ fn build_blob_array_descriptors(
Ok(BlobReadValue::Array(elements))
}
+async fn fetch_blob_map_values(
+ reader: &dyn FileRead,
+ planned_reads: Vec<PlannedBlobArrayRead>,
+ file_path: &str,
+ descriptor_mode: bool,
+) -> crate::Result<Vec<BlobReadValue>> {
+ futures::stream::iter(planned_reads.into_iter().map(|planned_read| async move {
+ match planned_read {
+ PlannedBlobArrayRead::Null => Ok(BlobReadValue::Null),
+ PlannedBlobArrayRead::Placeholder => Ok(BlobReadValue::Placeholder),
+ PlannedBlobArrayRead::Read(payload_range) => {
+ read_blob_map_entry(reader, payload_range, file_path, descriptor_mode).await
+ }
+ }
+ }))
+ .buffered(BLOB_READ_CONCURRENCY)
+ .try_collect()
+ .await
+}
+
+async fn read_blob_map_entry(
+ reader: &dyn FileRead,
+ payload_range: Range<u64>,
+ file_path: &str,
+ descriptor_mode: bool,
+) -> crate::Result<BlobReadValue> {
+ let payload_length = payload_range
+ .end
+ .checked_sub(payload_range.start)
+ .ok_or_else(|| Error::DataInvalid {
+ message: format!("Invalid MAP<X, BLOB> payload range: {payload_range:?}"),
+ source: None,
+ })?;
+ if payload_length < BLOB_MAP_MIN_PAYLOAD_SIZE {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "MAP<X, BLOB> payload is too small: expected at least {BLOB_MAP_MIN_PAYLOAD_SIZE} bytes, got {payload_length}"
+ ),
+ source: None,
+ });
+ }
+
+ let header = read_blob_map_range(
+ reader,
+ payload_range.start..payload_range.start + BLOB_MAP_HEADER_SIZE,
+ "header",
+ )
+ .await?;
+ let magic = i32::from_le_bytes(header[..4].try_into().unwrap());
+ if magic != BLOB_MAP_MAGIC_NUMBER {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Invalid MAP<X, BLOB> payload magic number: expected {BLOB_MAP_MAGIC_NUMBER}, got {magic}"
+ ),
+ source: None,
+ });
+ }
+ if header[4] != BLOB_MAP_VERSION {
+ return Err(Error::Unsupported {
+ message: format!(
+ "Unsupported MAP<X, BLOB> payload version: expected {BLOB_MAP_VERSION}, got {}",
+ header[4]
+ ),
+ });
+ }
+ let entry_count = i32::from_le_bytes(header[5..9].try_into().unwrap());
+ if entry_count < 0 {
+ return Err(Error::DataInvalid {
+ message: format!("Invalid MAP<X, BLOB> entry count: {entry_count}"),
+ source: None,
+ });
+ }
+ let entry_count = entry_count as usize;
+
+ let index_lengths_start = payload_range.end - BLOB_MAP_INDEX_LENGTHS_SIZE;
+ let index_lengths = read_blob_map_range(
+ reader,
+ index_lengths_start..payload_range.end,
+ "index lengths",
+ )
+ .await?;
+ let key_index_length = i32::from_le_bytes(index_lengths[..4].try_into().unwrap());
+ let value_index_length = i32::from_le_bytes(index_lengths[4..8].try_into().unwrap());
+ let max_indexes = payload_length - BLOB_MAP_MIN_PAYLOAD_SIZE;
+ if key_index_length < 0 || key_index_length as u64 > max_indexes {
+ return Err(Error::DataInvalid {
+ message: format!("Invalid MAP<X, BLOB> key index length: {key_index_length}"),
+ source: None,
+ });
+ }
+ if value_index_length < 0 || value_index_length as u64 > max_indexes {
+ return Err(Error::DataInvalid {
+ message: format!("Invalid MAP<X, BLOB> value index length: {value_index_length}"),
+ source: None,
+ });
+ }
+ let key_index_length = key_index_length as u64;
+ let value_index_length = value_index_length as u64;
+ if key_index_length + value_index_length > max_indexes
+ || entry_count as u64 > key_index_length
+ || entry_count as u64 > value_index_length
+ {
+ return Err(Error::DataInvalid {
+ message: "MAP<X, BLOB> indexes do not match the payload".to_string(),
+ source: None,
+ });
+ }
+
+ let value_index_start = index_lengths_start - value_index_length;
+ let key_index_start = value_index_start - key_index_length;
+ let key_index =
+ read_blob_map_range(reader, key_index_start..value_index_start, "key index").await?;
+ let value_index = read_blob_map_range(
+ reader,
+ value_index_start..index_lengths_start,
+ "value index",
+ )
+ .await?;
+ let key_lengths = decode_delta_varints(&key_index).map_err(|e| Error::DataInvalid {
+ message: format!("Invalid MAP<X, BLOB> key index: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ let value_lengths = decode_delta_varints(&value_index).map_err(|e| Error::DataInvalid {
+ message: format!("Invalid MAP<X, BLOB> value index: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ if key_lengths.len() != entry_count || value_lengths.len() != entry_count {
+ return Err(Error::DataInvalid {
+ message: "MAP<X, BLOB> entry count does not match index lengths".to_string(),
+ source: None,
+ });
+ }
+
+ let data_start = payload_range.start + BLOB_MAP_HEADER_SIZE;
+ let data_length = key_index_start - data_start;
+ let mut key_data_length = 0u64;
+ for &length in &key_lengths {
+ if length == BLOB_MAP_NULL_LENGTH {
+ return Err(Error::DataInvalid {
+ message: "MAP<X, BLOB> null keys cannot be represented by Arrow".to_string(),
+ source: None,
+ });
+ }
+ let length = u64::try_from(length).map_err(|e| Error::DataInvalid {
+ message: format!("Invalid MAP<X, BLOB> key length: {length}"),
+ source: Some(Box::new(e)),
+ })?;
+ key_data_length = key_data_length
+ .checked_add(length)
+ .filter(|total| *total <= data_length)
+ .ok_or_else(|| Error::DataInvalid {
+ message: "MAP<X, BLOB> key lengths exceed the payload data length".to_string(),
+ source: None,
+ })?;
+ }
+ let value_data_length = data_length - key_data_length;
+ let mut total_value_length = 0u64;
+ for &length in &value_lengths {
+ if length == BLOB_MAP_NULL_LENGTH {
+ continue;
+ }
+ let length = u64::try_from(length).map_err(|e| Error::DataInvalid {
+ message: format!("Invalid MAP<X, BLOB> value length: {length}"),
+ source: Some(Box::new(e)),
+ })?;
+ total_value_length = total_value_length
+ .checked_add(length)
+ .filter(|total| *total <= value_data_length)
+ .ok_or_else(|| Error::DataInvalid {
+ message: "MAP<X, BLOB> value lengths exceed the payload data length".to_string(),
+ source: None,
+ })?;
+ }
+ if total_value_length != value_data_length {
+ return Err(Error::DataInvalid {
+ message: "MAP<X, BLOB> key/value lengths do not match the payload data length"
+ .to_string(),
+ source: None,
+ });
+ }
+
+ let key_data =
+ read_blob_map_range(reader, data_start..data_start + key_data_length, "key data").await?;
+ let mut keys = Vec::with_capacity(entry_count);
+ let mut cursor = 0usize;
+ let mut unique = std::collections::HashSet::with_capacity(entry_count);
+ for length in key_lengths {
+ let length = length as usize;
+ let end = cursor + length;
+ let key = key_data.slice(cursor..end);
+ if !unique.insert(key.clone()) {
+ return Err(Error::DataInvalid {
+ message: "Invalid MAP<X, BLOB> payload: duplicate key".to_string(),
+ source: None,
+ });
+ }
+ keys.push(key);
+ cursor = end;
+ }
+
+ let mut value_offset = data_start + key_data_length;
+ let mut reads = Vec::with_capacity(entry_count);
+ for length in value_lengths {
+ if length == BLOB_MAP_NULL_LENGTH {
+ reads.push(None);
+ } else {
+ let length = length as u64;
+ reads.push(Some(value_offset..value_offset + length));
+ value_offset += length;
+ }
+ }
+ let values = if descriptor_mode {
+ reads
+ .into_iter()
+ .map(|range| {
+ range
+ .map(|range| {
+ let offset =
+ i64::try_from(range.start).map_err(|e| Error::DataInvalid {
+ message: "MAP<X, BLOB> descriptor offset exceeds i64".to_string(),
+ source: Some(Box::new(e)),
+ })?;
+ let length = i64::try_from(range.end - range.start).map_err(|e| {
+ Error::DataInvalid {
+ message: "MAP<X, BLOB> descriptor length exceeds i64".to_string(),
+ source: Some(Box::new(e)),
+ }
+ })?;
+ Ok(Bytes::from(
+ BlobDescriptor::new(file_path.to_string(), offset, length).serialize(),
+ ))
+ })
+ .transpose()
+ })
+ .collect::<crate::Result<Vec<_>>>()?
+ } else {
+ let payload = read_blob_entry(reader, blob_entry_range(&payload_range)).await?;
Review Comment:
Please reject oversized inline payloads before this read. total_value_length is a u64, but these values are later collected into an Arrow BinaryArray, whose 32-bit offsets panic once the total byte length exceeds i32::MAX. ARRAY<BLOB> already preflights this case; MAP should do the same and also validate the flattened batch total.
##########
bindings/go/tests/blob_reader_test.go:
##########
@@ -101,6 +104,61 @@ func TestBlobReaderReadBlobAndBatch(t *testing.T) {
}
}
+func TestStringBlobMapDescriptors(t *testing.T) {
Review Comment:
Could this test cover the actual read path instead of starting from a hand-built Arrow map? As written, it only tests descriptor extraction and BlobReader. A writer-generated fixture read through NewReadBuilderWithOptions would also cover the Rust decoder, C Data conversion, and Go API.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(blob): support MAP reads in Go [paimon-rust]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #784:
URL: https://github.com/apache/paimon-rust/pull/784
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(blob): support MAP reads in Go [paimon-rust]
Posted by "XiaoHongbo-Hope (via GitHub)" <gi...@apache.org>.
XiaoHongbo-Hope commented on code in PR #784:
URL: https://github.com/apache/paimon-rust/pull/784#discussion_r3915383771
##########
crates/paimon/src/arrow/format/blob.rs:
##########
@@ -764,6 +995,277 @@ fn build_blob_array_descriptors(
Ok(BlobReadValue::Array(elements))
}
+async fn fetch_blob_map_values(
+ reader: &dyn FileRead,
+ planned_reads: Vec<PlannedBlobArrayRead>,
+ file_path: &str,
+ descriptor_mode: bool,
+) -> crate::Result<Vec<BlobReadValue>> {
+ futures::stream::iter(planned_reads.into_iter().map(|planned_read| async move {
+ match planned_read {
+ PlannedBlobArrayRead::Null => Ok(BlobReadValue::Null),
+ PlannedBlobArrayRead::Placeholder => Ok(BlobReadValue::Placeholder),
+ PlannedBlobArrayRead::Read(payload_range) => {
+ read_blob_map_entry(reader, payload_range, file_path, descriptor_mode).await
+ }
+ }
+ }))
+ .buffered(BLOB_READ_CONCURRENCY)
+ .try_collect()
+ .await
+}
+
+async fn read_blob_map_entry(
+ reader: &dyn FileRead,
+ payload_range: Range<u64>,
+ file_path: &str,
+ descriptor_mode: bool,
+) -> crate::Result<BlobReadValue> {
+ let payload_length = payload_range
+ .end
+ .checked_sub(payload_range.start)
+ .ok_or_else(|| Error::DataInvalid {
+ message: format!("Invalid MAP<X, BLOB> payload range: {payload_range:?}"),
+ source: None,
+ })?;
+ if payload_length < BLOB_MAP_MIN_PAYLOAD_SIZE {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "MAP<X, BLOB> payload is too small: expected at least {BLOB_MAP_MIN_PAYLOAD_SIZE} bytes, got {payload_length}"
+ ),
+ source: None,
+ });
+ }
+
+ let header = read_blob_map_range(
+ reader,
+ payload_range.start..payload_range.start + BLOB_MAP_HEADER_SIZE,
+ "header",
+ )
+ .await?;
+ let magic = i32::from_le_bytes(header[..4].try_into().unwrap());
+ if magic != BLOB_MAP_MAGIC_NUMBER {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Invalid MAP<X, BLOB> payload magic number: expected {BLOB_MAP_MAGIC_NUMBER}, got {magic}"
+ ),
+ source: None,
+ });
+ }
+ if header[4] != BLOB_MAP_VERSION {
+ return Err(Error::Unsupported {
+ message: format!(
+ "Unsupported MAP<X, BLOB> payload version: expected {BLOB_MAP_VERSION}, got {}",
+ header[4]
+ ),
+ });
+ }
+ let entry_count = i32::from_le_bytes(header[5..9].try_into().unwrap());
+ if entry_count < 0 {
+ return Err(Error::DataInvalid {
+ message: format!("Invalid MAP<X, BLOB> entry count: {entry_count}"),
+ source: None,
+ });
+ }
+ let entry_count = entry_count as usize;
+
+ let index_lengths_start = payload_range.end - BLOB_MAP_INDEX_LENGTHS_SIZE;
+ let index_lengths = read_blob_map_range(
+ reader,
+ index_lengths_start..payload_range.end,
+ "index lengths",
+ )
+ .await?;
+ let key_index_length = i32::from_le_bytes(index_lengths[..4].try_into().unwrap());
+ let value_index_length = i32::from_le_bytes(index_lengths[4..8].try_into().unwrap());
+ let max_indexes = payload_length - BLOB_MAP_MIN_PAYLOAD_SIZE;
+ if key_index_length < 0 || key_index_length as u64 > max_indexes {
+ return Err(Error::DataInvalid {
+ message: format!("Invalid MAP<X, BLOB> key index length: {key_index_length}"),
+ source: None,
+ });
+ }
+ if value_index_length < 0 || value_index_length as u64 > max_indexes {
+ return Err(Error::DataInvalid {
+ message: format!("Invalid MAP<X, BLOB> value index length: {value_index_length}"),
+ source: None,
+ });
+ }
+ let key_index_length = key_index_length as u64;
+ let value_index_length = value_index_length as u64;
+ if key_index_length + value_index_length > max_indexes
+ || entry_count as u64 > key_index_length
+ || entry_count as u64 > value_index_length
+ {
+ return Err(Error::DataInvalid {
+ message: "MAP<X, BLOB> indexes do not match the payload".to_string(),
+ source: None,
+ });
+ }
+
+ let value_index_start = index_lengths_start - value_index_length;
+ let key_index_start = value_index_start - key_index_length;
+ let key_index =
+ read_blob_map_range(reader, key_index_start..value_index_start, "key index").await?;
+ let value_index = read_blob_map_range(
+ reader,
+ value_index_start..index_lengths_start,
+ "value index",
+ )
+ .await?;
+ let key_lengths = decode_delta_varints(&key_index).map_err(|e| Error::DataInvalid {
+ message: format!("Invalid MAP<X, BLOB> key index: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ let value_lengths = decode_delta_varints(&value_index).map_err(|e| Error::DataInvalid {
+ message: format!("Invalid MAP<X, BLOB> value index: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ if key_lengths.len() != entry_count || value_lengths.len() != entry_count {
+ return Err(Error::DataInvalid {
+ message: "MAP<X, BLOB> entry count does not match index lengths".to_string(),
+ source: None,
+ });
+ }
+
+ let data_start = payload_range.start + BLOB_MAP_HEADER_SIZE;
+ let data_length = key_index_start - data_start;
+ let mut key_data_length = 0u64;
+ for &length in &key_lengths {
+ if length == BLOB_MAP_NULL_LENGTH {
+ return Err(Error::DataInvalid {
+ message: "MAP<X, BLOB> null keys cannot be represented by Arrow".to_string(),
+ source: None,
+ });
+ }
+ let length = u64::try_from(length).map_err(|e| Error::DataInvalid {
+ message: format!("Invalid MAP<X, BLOB> key length: {length}"),
+ source: Some(Box::new(e)),
+ })?;
+ key_data_length = key_data_length
+ .checked_add(length)
+ .filter(|total| *total <= data_length)
+ .ok_or_else(|| Error::DataInvalid {
+ message: "MAP<X, BLOB> key lengths exceed the payload data length".to_string(),
+ source: None,
+ })?;
+ }
+ let value_data_length = data_length - key_data_length;
+ let mut total_value_length = 0u64;
+ for &length in &value_lengths {
+ if length == BLOB_MAP_NULL_LENGTH {
+ continue;
+ }
+ let length = u64::try_from(length).map_err(|e| Error::DataInvalid {
+ message: format!("Invalid MAP<X, BLOB> value length: {length}"),
+ source: Some(Box::new(e)),
+ })?;
+ total_value_length = total_value_length
+ .checked_add(length)
+ .filter(|total| *total <= value_data_length)
+ .ok_or_else(|| Error::DataInvalid {
+ message: "MAP<X, BLOB> value lengths exceed the payload data length".to_string(),
+ source: None,
+ })?;
+ }
+ if total_value_length != value_data_length {
+ return Err(Error::DataInvalid {
+ message: "MAP<X, BLOB> key/value lengths do not match the payload data length"
+ .to_string(),
+ source: None,
+ });
+ }
+
+ let key_data =
+ read_blob_map_range(reader, data_start..data_start + key_data_length, "key data").await?;
+ let mut keys = Vec::with_capacity(entry_count);
+ let mut cursor = 0usize;
+ let mut unique = std::collections::HashSet::with_capacity(entry_count);
+ for length in key_lengths {
+ let length = length as usize;
+ let end = cursor + length;
+ let key = key_data.slice(cursor..end);
+ if !unique.insert(key.clone()) {
+ return Err(Error::DataInvalid {
+ message: "Invalid MAP<X, BLOB> payload: duplicate key".to_string(),
+ source: None,
+ });
+ }
+ keys.push(key);
+ cursor = end;
+ }
+
+ let mut value_offset = data_start + key_data_length;
+ let mut reads = Vec::with_capacity(entry_count);
+ for length in value_lengths {
+ if length == BLOB_MAP_NULL_LENGTH {
+ reads.push(None);
+ } else {
+ let length = length as u64;
+ reads.push(Some(value_offset..value_offset + length));
+ value_offset += length;
+ }
+ }
+ let values = if descriptor_mode {
+ reads
+ .into_iter()
+ .map(|range| {
+ range
+ .map(|range| {
+ let offset =
+ i64::try_from(range.start).map_err(|e| Error::DataInvalid {
+ message: "MAP<X, BLOB> descriptor offset exceeds i64".to_string(),
+ source: Some(Box::new(e)),
+ })?;
+ let length = i64::try_from(range.end - range.start).map_err(|e| {
+ Error::DataInvalid {
+ message: "MAP<X, BLOB> descriptor length exceeds i64".to_string(),
+ source: Some(Box::new(e)),
+ }
+ })?;
+ Ok(Bytes::from(
+ BlobDescriptor::new(file_path.to_string(), offset, length).serialize(),
+ ))
+ })
+ .transpose()
+ })
+ .collect::<crate::Result<Vec<_>>>()?
+ } else {
+ let payload = read_blob_entry(reader, blob_entry_range(&payload_range)).await?;
Review Comment:
Fixed in cd19225. Inline MAP values are now rejected before the complete entry read, and flattened key/value batches enforce Arrow i32 offset limits. Added sparse >2 GiB regression coverage.
##########
bindings/go/tests/blob_reader_test.go:
##########
@@ -101,6 +104,61 @@ func TestBlobReaderReadBlobAndBatch(t *testing.T) {
}
}
+func TestStringBlobMapDescriptors(t *testing.T) {
Review Comment:
Fixed in cd19225. The Go test now reads a PyPaimon-generated table through NewReadBuilderWithOptions, Rust decoding, C Data import, Go map extraction, and ReadBlobs.
##########
docs/src/go-binding.md:
##########
@@ -147,8 +147,8 @@ catalog, err := paimon.NewCatalog(map[string]string{
"token.provider": "dlf",
"dlf.region": os.Getenv("DLF_REGION"),
"dlf.oss-endpoint": os.Getenv("DLF_OSS_ENDPOINT"),
- "dlf.token-loader": "ecs",
- "dlf.token-ecs-role-name": os.Getenv("DLF_ECS_ROLE"),
+ "dlf.access-key-id": os.Getenv("DLF_ACCESS_KEY_ID"),
Review Comment:
Fixed in cd19225. Restored the ECS RAM-role loader example with dlf.token-loader and dlf.token-ecs-role-name.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org