Reject changelog-producer on non-PK tables and decouple changelog file format from data file format

Author: junmuz

paimon-python/pypaimon/common/options/core_options.py

@@ -292,6 +292,14 @@ class CoreOptions:
                           "Options: none, input, full-compaction, lookup.")
     )
 
+    CHANGELOG_FILE_FORMAT: ConfigOption[str] = (
+        ConfigOptions.key("changelog-file.format")
+        .string_type()
+        .no_default_value()
+        .with_description("Specify the file format of changelog files. "
+                          "Currently parquet, avro and orc are supported.")
+    )
+
     MERGE_ENGINE: ConfigOption[MergeEngine] = (
         ConfigOptions.key("merge-engine")
         .enum_type(MergeEngine)
@@ -542,6 +550,9 @@ def deletion_vectors_enabled(self, default=None):
     def changelog_producer(self, default=None):
         return self.options.get(CoreOptions.CHANGELOG_PRODUCER, default)
 
+    def changelog_file_format(self, default=None):
+        return self.options.get(CoreOptions.CHANGELOG_FILE_FORMAT, default)
+
     def merge_engine(self, default=None):
         return self.options.get(CoreOptions.MERGE_ENGINE, default)
 

paimon-python/pypaimon/schema/schema.py

@@ -92,4 +92,11 @@ def from_pyarrow_schema(pa_schema: pa.Schema, partition_keys: Optional[List[str]
             if primary_keys is not None:
                 raise ValueError("Blob type is not supported with primary key.")
 
+        changelog_producer = (options or {}).get(CoreOptions.CHANGELOG_PRODUCER.key(), 'none')
+        if changelog_producer != 'none' and not primary_keys:
+            raise ValueError(
+                f"Cannot set 'changelog-producer' to '{changelog_producer}' on a table without primary keys. "
+                f"Changelog producer requires primary keys to be defined."
+            )
+
         return Schema(fields, partition_keys, primary_keys, options, comment)

paimon-python/pypaimon/tests/write/changelog_producer_test.py

@@ -266,6 +266,94 @@ def test_abort_cleans_up_changelog_files(self):
         table_write.close()
         table_commit.close()
 
+    def test_reject_changelog_producer_on_append_only_table(self):
+        append_schema = pa.schema([
+            ('user_id', pa.int32()),
+            ('item_id', pa.int64()),
+            ('behavior', pa.string()),
+            ('dt', pa.string())
+        ])
+        for mode in ['input', 'full-compaction', 'lookup']:
+            with self.assertRaises(ValueError, msg=f"Should reject changelog-producer={mode} without PKs"):
+                Schema.from_pyarrow_schema(
+                    append_schema,
+                    partition_keys=['dt'],
+                    options={'changelog-producer': mode, 'bucket': '1'}
+                )
+
+    def test_changelog_producer_none_allowed_on_append_only_table(self):
+        append_schema = pa.schema([
+            ('user_id', pa.int32()),
+            ('item_id', pa.int64()),
+            ('behavior', pa.string()),
+            ('dt', pa.string())
+        ])
+        schema = Schema.from_pyarrow_schema(
+            append_schema,
+            partition_keys=['dt'],
+            options={'changelog-producer': 'none', 'bucket': '1'}
+        )
+        self.assertIsNotNone(schema)
+
+    def test_input_mode_changelog_uses_parquet_regardless_of_data_format(self):
+        table = self._create_table(
+            'test_input_changelog_format',
+            options={'changelog-producer': 'input', 'bucket': '1', 'file.format': 'orc'}
+        )
+        write_builder = table.new_batch_write_builder()
+        table_write = write_builder.new_write()
+        table_commit = write_builder.new_commit()
+
+        table_write.write_arrow(self._sample_data())
+        table_commit.commit(table_write.prepare_commit())
+
+        bucket_dir = os.path.join(
+            self.warehouse, 'default.db', 'test_input_changelog_format', 'dt=p1', 'bucket-0')
+        changelog_files = glob.glob(os.path.join(bucket_dir, 'changelog-*'))
+        self.assertTrue(len(changelog_files) > 0, "Should have changelog files")
+        for f in changelog_files:
+            self.assertTrue(f.endswith('.parquet'),
+                            f"Changelog file should use parquet format by default, got {f}")
+
+        data_files = glob.glob(os.path.join(bucket_dir, 'data-*'))
+        self.assertTrue(len(data_files) > 0, "Should have data files")
+        for f in data_files:
+            self.assertTrue(f.endswith('.orc'),
+                            f"Data file should use orc format, got {f}")
+
+        table_write.close()
+        table_commit.close()
+
+    def test_input_mode_changelog_respects_changelog_file_format(self):
+        table = self._create_table(
+            'test_input_cl_file_fmt',
+            options={'changelog-producer': 'input', 'bucket': '1',
+                     'file.format': 'parquet', 'changelog-file.format': 'orc'}
+        )
+        write_builder = table.new_batch_write_builder()
+        table_write = write_builder.new_write()
+        table_commit = write_builder.new_commit()
+
+        table_write.write_arrow(self._sample_data())
+        table_commit.commit(table_write.prepare_commit())
+
+        bucket_dir = os.path.join(
+            self.warehouse, 'default.db', 'test_input_cl_file_fmt', 'dt=p1', 'bucket-0')
+        changelog_files = glob.glob(os.path.join(bucket_dir, 'changelog-*'))
+        self.assertTrue(len(changelog_files) > 0, "Should have changelog files")
+        for f in changelog_files:
+            self.assertTrue(f.endswith('.orc'),
+                            f"Changelog file should use orc format, got {f}")
+
+        data_files = glob.glob(os.path.join(bucket_dir, 'data-*'))
+        self.assertTrue(len(data_files) > 0, "Should have data files")
+        for f in data_files:
+            self.assertTrue(f.endswith('.parquet'),
+                            f"Data file should use parquet format, got {f}")
+
+        table_write.close()
+        table_commit.close()
+
 
 if __name__ == '__main__':
     unittest.main()

paimon-python/pypaimon/write/file_store_commit.py

@@ -127,7 +127,6 @@ def commit(self, commit_messages: List[CommitMessage], commit_identifier: int):
             len(commit_messages),
         )
         commit_entries = []
-        changelog_entries = []
         for msg in commit_messages:
             partition = GenericRow(list(msg.partition), self.table.partition_keys_fields)
             for file in msg.new_files:
@@ -138,17 +137,10 @@ def commit(self, commit_messages: List[CommitMessage], commit_identifier: int):
                     total_buckets=self.table.total_buckets,
                     file=file
                 ))
-            for file in msg.changelog_files:
-                changelog_entries.append(ManifestEntry(
-                    kind=0,
-                    partition=partition,
-                    bucket=msg.bucket,
-                    total_buckets=self.table.total_buckets,
-                    file=file
-                ))
+        changelog_entries = self._collect_changelog_entries(commit_messages)
 
         logger.info("Finished collecting changes, including: %d entries, %d changelog entries",
-                     len(commit_entries), len(changelog_entries))
+                    len(commit_entries), len(changelog_entries))
 
         commit_kind = "APPEND"
         detect_conflicts = False
@@ -193,17 +185,7 @@ def overwrite(self, overwrite_partition, commit_messages: List[CommitMessage], c
                     raise RuntimeError(f"Trying to overwrite partition {overwrite_partition}, but the changes "
                                        f"in {msg.partition} does not belong to this partition")
 
-        changelog_entries = []
-        for msg in commit_messages:
-            partition = GenericRow(list(msg.partition), self.table.partition_keys_fields)
-            for file in msg.changelog_files:
-                changelog_entries.append(ManifestEntry(
-                    kind=0,
-                    partition=partition,
-                    bucket=msg.bucket,
-                    total_buckets=self.table.total_buckets,
-                    file=file
-                ))
+        changelog_entries = self._collect_changelog_entries(commit_messages)
 
         self._try_commit(
             commit_kind="OVERWRITE",
@@ -600,6 +582,20 @@ def _commit_retry_wait(self, retry_count: int):
 
         time.sleep(total_wait_ms / 1000.0)
 
+    def _collect_changelog_entries(self, commit_messages: List[CommitMessage]) -> List[ManifestEntry]:
+        changelog_entries = []
+        for msg in commit_messages:
+            partition = GenericRow(list(msg.partition), self.table.partition_keys_fields)
+            for file in msg.changelog_files:
+                changelog_entries.append(ManifestEntry(
+                    kind=0,
+                    partition=partition,
+                    bucket=msg.bucket,
+                    total_buckets=self.table.total_buckets,
+                    file=file
+                ))
+        return changelog_entries
+
     def _cleanup_preparation_failure(self,
                                      delta_manifest_list: Optional[str],
                                      base_manifest_list: Optional[str],

paimon-python/pypaimon/write/writer/data_writer.py

@@ -64,6 +64,10 @@ def __init__(self, table, partition: Tuple, bucket: int, max_seq_number: int, op
         self.committed_files: List[DataFileMeta] = []
         self.committed_changelog_files: List[DataFileMeta] = []
         self.changelog_producer = changelog_producer
+        self.changelog_file_format = (
+            self.options.changelog_file_format()
+            or CoreOptions.FILE_FORMAT_PARQUET
+        )
         self.write_cols = write_cols
         self.blob_as_descriptor = self.options.blob_as_descriptor()
 
@@ -252,25 +256,26 @@ def _write_data_to_file(self, data: pa.Table):
     def _write_changelog_file(self, data, min_key, max_key, key_stats, value_stats,
                               min_seq, max_seq, creation_time,
                               value_stats_enabled, is_external):
-        changelog_file_name = f"changelog-{uuid.uuid4()}-0.{self.file_format}"
+        cl_fmt = self.changelog_file_format
+        changelog_file_name = f"changelog-{uuid.uuid4()}-0.{cl_fmt}"
         changelog_file_path = self._generate_file_path(changelog_file_name)
 
         changelog_external_path = None
         if is_external and self._current_external_path:
             changelog_external_path = self._current_external_path
 
-        if self.file_format == CoreOptions.FILE_FORMAT_PARQUET:
+        if cl_fmt == CoreOptions.FILE_FORMAT_PARQUET:
             self.file_io.write_parquet(changelog_file_path, data, compression=self.compression,
                                        zstd_level=self.zstd_level)
-        elif self.file_format == CoreOptions.FILE_FORMAT_ORC:
+        elif cl_fmt == CoreOptions.FILE_FORMAT_ORC:
             self.file_io.write_orc(changelog_file_path, data, compression=self.compression,
-                                    zstd_level=self.zstd_level)
-        elif self.file_format == CoreOptions.FILE_FORMAT_AVRO:
+                                   zstd_level=self.zstd_level)
+        elif cl_fmt == CoreOptions.FILE_FORMAT_AVRO:
             self.file_io.write_avro(changelog_file_path, data, compression=self.compression,
-                                     zstd_level=self.zstd_level)
+                                    zstd_level=self.zstd_level)
         else:
-            self.file_io.write_parquet(changelog_file_path, data, compression=self.compression,
-                                       zstd_level=self.zstd_level)
+            raise ValueError(f"Unsupported changelog file format: {cl_fmt}. "
+                             f"Supported formats: parquet, orc, avro.")
 
         self.committed_changelog_files.append(DataFileMeta.create(
             file_name=changelog_file_name,