feat(write): support INSERT INTO for Paimon tables

Implement PlanInsert in the Paimon catalog to enable appending data to
existing Paimon tables via INSERT INTO ... SELECT. The implementation
reuses the existing parallel sink operator in insert-only mode (no table
creation), supporting append-only tables.

Also fix a bug where primary key constraints were read from the empty
BoundCreateTableInfo.constraints instead of the populated
CreateTableInfo.constraints, causing PRIMARY KEY clauses to be silently
ignored during table creation.

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>

Author: mrdrivingduck

README.md

@@ -20,6 +20,8 @@ This extension is built on top of [paimon-cpp](https://github.com/alibaba/paimon
 ## Features
 
 - Read Paimon table data (local and remote OSS)
+- Write Paimon tables (CREATE TABLE AS, INSERT INTO; append-only tables only)
+- DDL support (CREATE/DROP SCHEMA, CREATE/DROP TABLE)
 - Projection pushdown optimization
 - Predicate pushdown optimization
 - Multiple file format support (manifest / data)
@@ -151,6 +153,35 @@ SELECT * FROM my_catalog.testdb.testtbl;
 -- ATTACH 'oss://my-bucket/warehouse' AS my_catalog (TYPE paimon);
 ```
 
+### Write Data
+
+With an ATTACHed catalog, you can create schemas, create tables, and insert data using standard DuckDB SQL:
+
+```sql
+ATTACH './data' AS my_catalog (TYPE paimon);
+
+CREATE SCHEMA my_catalog.my_db;
+
+CREATE TABLE my_catalog.my_db.orders AS
+    SELECT 1 AS order_id, 99.9::DECIMAL(18,2) AS amount, 'Alice' AS customer;
+
+INSERT INTO my_catalog.my_db.orders
+    SELECT 2, 49.5, 'Bob'
+    UNION ALL
+    SELECT 3, 150.0, 'Charlie';
+
+SELECT * FROM my_catalog.my_db.orders ORDER BY order_id;
+
+DROP TABLE my_catalog.my_db.orders;
+DROP SCHEMA my_catalog.my_db;
+```
+
+To prevent accidental writes, attach the catalog in read-only mode:
+
+```sql
+ATTACH './data' AS my_catalog (TYPE paimon, READ_ONLY);
+```
+
 ### Inspect Snapshot History
 
 Use `paimon_snapshots` to list all snapshots of a Paimon table — useful for auditing commit history, diagnosing data issues, or identifying a snapshot ID for time-travel queries:

src/paimon_storage/paimon_catalog.cpp

@@ -23,6 +23,7 @@
  */
 
 #include "duckdb/catalog/catalog.hpp"
+#include "duckdb/catalog/catalog_entry/table_catalog_entry.hpp"
 #include "duckdb/common/string_util.hpp"
 #include "duckdb/common/types/timestamp.hpp"
 #include "duckdb/main/secret/secret_manager.hpp"
@@ -31,6 +32,7 @@
 #include "duckdb/parser/parsed_data/attach_info.hpp"
 #include "duckdb/parser/parsed_data/create_table_info.hpp"
 #include "duckdb/planner/operator/logical_create_table.hpp"
+#include "duckdb/planner/operator/logical_insert.hpp"
 #include "duckdb/planner/parsed_data/bound_create_table_info.hpp"
 
 #include "paimon/catalog/catalog.h"
@@ -240,9 +242,26 @@ PhysicalOperator &PaimonCatalog::PlanCreateTableAs(ClientContext &context, Physi
 	return insert;
 }
 
-PhysicalOperator &PaimonCatalog::PlanInsert(ClientContext &, PhysicalPlanGenerator &, LogicalInsert &,
-                                            optional_ptr<PhysicalOperator>) {
-	throw NotImplementedException("PlanInsert not supported yet");
+PhysicalOperator &PaimonCatalog::PlanInsert(ClientContext &context, PhysicalPlanGenerator &planner, LogicalInsert &op,
+                                            optional_ptr<PhysicalOperator> plan) {
+	if (access_mode == AccessMode::READ_ONLY) {
+		throw PermissionException("Cannot write to a read-only Paimon catalog");
+	}
+
+	auto &table = op.table;
+	auto table_path = path + "/" + table.schema.name + paimon::Catalog::DB_SUFFIX + "/" + table.name;
+	auto paimon_options = GetPaimonOptions(context, path, attached_options);
+
+	if (plan && !op.column_index_map.empty()) {
+		plan = planner.ResolveDefaultsProjection(op, *plan);
+	}
+
+	auto &insert = planner.Make<PhysicalPaimonInsert>(op, table.schema, nullptr, std::move(table_path),
+	                                                  std::move(paimon_options), 0U);
+	if (plan) {
+		insert.children.push_back(*plan);
+	}
+	return insert;
 }
 
 PhysicalOperator &PaimonCatalog::PlanDelete(ClientContext &, PhysicalPlanGenerator &, LogicalDelete &,

src/paimon_storage/paimon_schema_entry.cpp

@@ -111,7 +111,7 @@ optional_ptr<CatalogEntry> PaimonSchemaEntry::CreateTable(CatalogTransaction tra
 	} arrow_guard {arrow_schema};
 
 	vector<string> primary_keys;
-	for (auto &constraint : info.constraints) {
+	for (auto &constraint : base.constraints) {
 		if (constraint->type != ConstraintType::UNIQUE) {
 			continue;
 		}

test/sql/paimon_write.test

@@ -71,7 +71,71 @@ SELECT min(id), max(id) FROM pm.__test_write.t_ctas_large;
 ----
 1	100000
 
-# --- CTAS into read-only catalog should fail ---
+# --- INSERT INTO existing table ---
+
+statement ok
+INSERT INTO pm.__test_write.t_ctas_small
+    SELECT * FROM (VALUES
+        (4, 400000000000::BIGINT, 'duck', 6.28::FLOAT, 1.414213562::DOUBLE, FALSE, '2025-12-25'::DATE, '2025-12-25 00:00:00'::TIMESTAMP),
+        (5, 500000000000::BIGINT, 'paimon', 9.81::FLOAT, 2.236067977::DOUBLE, TRUE, '2025-07-04'::DATE, '2025-07-04 12:00:00'::TIMESTAMP)
+    ) t(id, big_id, name, score_f, score_d, flag, dt, ts);
+
+query I
+SELECT count(*) FROM pm.__test_write.t_ctas_small;
+----
+5
+
+query IITRRTTT
+SELECT * FROM pm.__test_write.t_ctas_small ORDER BY id;
+----
+1	100000000000	hello	3.14	2.718281828	true	2025-01-15	2025-01-15 10:30:00
+2	200000000000	world	1.41	3.141592653	false	2025-06-30	2025-06-30 23:59:59
+3	NULL	NULL	NULL	NULL	true	NULL	NULL
+4	400000000000	duck	6.28	1.414213562	false	2025-12-25	2025-12-25 00:00:00
+5	500000000000	paimon	9.81	2.236067977	true	2025-07-04	2025-07-04 12:00:00
+
+# --- INSERT INTO with large dataset ---
+
+statement ok
+INSERT INTO pm.__test_write.t_ctas_large
+    SELECT
+        i AS id,
+        i * 1000000000::BIGINT AS big_id,
+        'ins_' || i::VARCHAR AS name,
+        (i * 0.2)::FLOAT AS score_f,
+        i * 0.002 AS score_d,
+        (i % 2 = 1) AS flag,
+        ('2021-01-01'::DATE + (i % 500)::INTEGER)::DATE AS dt,
+        ('2021-01-01 00:00:00'::TIMESTAMP + INTERVAL (i) SECOND) AS ts
+    FROM generate_series(100001, 150000) t(i);
+
+query I
+SELECT count(*) FROM pm.__test_write.t_ctas_large;
+----
+150000
+
+query II
+SELECT min(id), max(id) FROM pm.__test_write.t_ctas_large;
+----
+1	150000
+
+# --- INSERT INTO with reordered / partial columns ---
+
+statement ok
+INSERT INTO pm.__test_write.t_ctas_small (name, id, flag, big_id, score_f, score_d, dt, ts)
+    VALUES ('reordered', 6, FALSE, 600000000000, 2.22, 0.123456789, '2025-03-15', '2025-03-15 08:00:00');
+
+statement ok
+INSERT INTO pm.__test_write.t_ctas_small (id, name)
+    VALUES (7, 'partial');
+
+query IITRRTTT
+SELECT * FROM pm.__test_write.t_ctas_small WHERE id >= 6 ORDER BY id;
+----
+6	600000000000	reordered	2.22	0.123456789	false	2025-03-15	2025-03-15 08:00:00
+7	NULL	partial	NULL	NULL	NULL	NULL	NULL
+
+# --- INSERT/CTAS into read-only catalog should fail ---
 
 statement ok
 DETACH pm;
@@ -84,6 +148,11 @@ CREATE TABLE pm_ro.__test_write.t_fail AS SELECT 1 AS id;
 ----
 read-only
 
+statement error
+INSERT INTO pm_ro.__test_write.t_ctas_small SELECT 1, 1::BIGINT, 'x', 1.0::FLOAT, 1.0, true, '2025-01-01'::DATE, '2025-01-01 00:00:00'::TIMESTAMP;
+----
+read-only
+
 statement ok
 DETACH pm_ro;