TheR1sing3un commented on code in PR #9798:
URL: https://github.com/apache/paimon/pull/9798#discussion_r4011757196
##########
paimon-python/pypaimon/globalindex/create_global_index.py:
##########
@@ -288,72 +289,164 @@ def _create_sorted_index_writer(self, index_path: str,
key_serializer):
def _build_generic_index(
self, splits, unindexed_ranges, index_field, table_read, index_path:
str
) -> List[CommitMessage]:
- from pypaimon.read.table_read import _ClosableArrowBatchReader
-
rows_per_shard = self._core_options.global_index_row_count_per_shard()
if rows_per_shard <= 0:
raise ValueError(
"Option 'global-index.row-count-per-shard' must be greater
than 0."
)
- messages = []
- for index_split, index_range in _split_by_global_index_shard(
- splits, rows_per_shard, unindexed_ranges
- ):
- writer = None
+ parallelism = self._core_options.global_index_build_parallelism()
+ if parallelism <= 0:
+ raise ValueError(
+ "Option 'global-index.build.parallelism' must be greater than
0."
+ )
+
+ shards = _split_by_global_index_shard(
+ splits, rows_per_shard, unindexed_ranges)
+ if not shards:
+ return []
+
+ if parallelism == 1 or len(shards) == 1:
+ messages = []
try:
- reader, batches =
table_read._new_arrow_batch_reader([index_split])
- # Close the Python iterator explicitly on failure as well as
- # the Arrow reader, which may retain a suspended generator.
- with _ClosableArrowBatchReader(reader, batches) as
batch_reader:
- for batch in batch_reader:
- if batch.num_rows == 0:
- continue
- if writer is None:
- writer = self._create_generic_index_writer(
- index_path, index_field)
- if self._index_type in VINDEX_IDENTIFIERS:
- if
batch.column(SpecialFields.ROW_ID.name).null_count:
- raise ValueError(
- "Cannot build global index because _ROW_ID
is null.")
- for offset in range(0, batch.num_rows,
ADD_BATCH_SIZE):
- _write_vector_batch(
- writer, batch.slice(offset,
ADD_BATCH_SIZE),
- self._index_columns[0], index_range)
- else:
- for value, row_id in _extract_index_rows(
- batch,
- self._index_columns[0],
- SpecialFields.ROW_ID.name,
- index_range,
- ):
- writer.write(value, row_id - index_range.from_)
- del batch
-
- if writer is None:
+ for index_split, index_range in shards:
+ message = self._build_generic_shard(
+ index_split, index_range, index_field, table_read,
index_path)
+ if message is not None:
+ messages.append(message)
+ return messages
+ except BaseException:
+ self._delete_uncommitted_indexes(messages)
+ raise
+
+ futures = []
+ try:
+ with ThreadPoolExecutor(
+ max_workers=min(parallelism, len(shards)),
+ thread_name_prefix="paimon-global-index-build",
+ ) as executor:
+ futures = [
Review Comment:
> [P2] Preserve output ownership when submitting a shard fails
Thanks, confirmed and fixed in e40b2bde3.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
