Do not add a partition spec if it already exists (#79)

Both Spark and Polaris follows this. Before this commit, we blindly added partition specs, even if the same spec existed before. Now, we are switching to a more proper way: if the same spec exists in the table earlier, we simply use that instead of adding one more spec.

This is especially useful for #68, where Polaris throws an error if we try to send a partition spec that already exists. Here `the same spec` means a spec that has the exact same fields with an existing spec in the table.

To give an example, previously we created 3 specs for the following, now 2:
```sql
ALTER TABLE t OPTION (ADD partition_by='bucket(10,a)');
..
ALTER TABLE t OPTION (ADD partition_by='bucket(20,a)');
..
-- back to 10, this does not generate a new spec anymore
ALTER TABLE t OPTION (ADD partition_by='bucket(10,a)');
```

On Polaris, before this commit, we'd get the following, given Polaris thinks this spec already exists:
```
alter table t (set partition_by 'bucket(10,a)');
WARNING:  HTTP request failed (HTTP 400)
DETAIL:  Cannot set last added spec: no spec has been added
HINT:  The rest catalog returned error type: ValidationException
```

Signed-off-by: Onder KALACI <[email protected]>

Author: sfc-gh-okalaci

pg_lake_table/include/pg_lake/fdw/partition_transform.h

@@ -28,6 +28,7 @@ extern void *ApplyBucketTransformToColumn(IcebergPartitionTransform * transform,
 										  Datum columnValue, bool isNull,
 										  size_t *bucketSize);
 extern List *CurrentPartitionTransformList(Oid relationId);
+extern IcebergPartitionSpec * GetPartitionSpecIfAlreadyExist(Oid relationId, List *partitionTransforms);
 extern List *AllPartitionTransformList(Oid relationId);
 extern List *GetPartitionTransformsFromSpecFields(Oid relationId, List *specFields);
 extern void *DeserializePartitionValueFromPGText(IcebergPartitionTransform * transform,

pg_lake_table/include/pg_lake/partitioning/partition_spec_catalog.h

@@ -43,6 +43,7 @@ typedef struct IcebergPartitionSpecHashEntry
 extern void UpdateDefaultPartitionSpecId(Oid relationId, int specId);
 extern void InsertPartitionSpecAndPartitionFields(Oid relationId, IcebergPartitionSpec * spec);
 extern int	GetLargestSpecId(Oid relationId);
+extern List *GetAllIcebergPartitionSpecIds(Oid relationId);
 extern PGDLLEXPORT int GetCurrentSpecId(Oid relationId);
 extern int	GetLargestPartitionFieldId(Oid relationId);
 extern IcebergPartitionSpecField * GetIcebergPartitionFieldFromCatalog(Oid relationId, int fieldId);

pg_lake_table/src/ddl/ddl_changes.c

@@ -212,6 +212,15 @@ ApplyDDLCatalogChanges(Oid relationId, List *ddlOperations,
 			List	   *parsedTransforms = ddlOperation->parsedTransforms;
 			List	   *analyzedTransforms = AnalyzeIcebergTablePartitionBy(relationId, parsedTransforms);
 
+			*partitionSpec = GetPartitionSpecIfAlreadyExist(relationId, analyzedTransforms);
+			if (*partitionSpec != NULL)
+			{
+				/* update lake_iceberg.internal_tables specId */
+				UpdateDefaultPartitionSpecId(relationId, (*partitionSpec)->spec_id);
+
+				return;
+			}
+
 			/*
 			 * creatint a new spec, get the largest spec and we'll increment
 			 * it if needed when assigning

pg_lake_table/src/fdw/partition_transform.c

@@ -63,6 +63,7 @@ static void ParseTransformName(const char *name, IcebergPartitionTransformType *
 static bool ParseBracketUintSize(const char *name, const char *prefix, size_t *outVal);
 static void *DatumToPartitionValue(IcebergPartitionTransform * transform, Datum columnValue, bool isNull,
 								   size_t *valuelength);
+static bool PartitionTransformsEqual(IcebergPartitionSpec * spec, List *partitionTransforms);
 
 /*
  * ComputePartitionTupleForTuple applies relative partition transforms
@@ -91,6 +92,84 @@ ComputePartitionTupleForTuple(List *transforms, TupleTableSlot *slot)
 }
 
 
+/*
+* If a partition spec with the given partition transforms already exists
+* for the given relation, returns the partition spec. Otherwise, it
+* returns NULL.
+*/
+IcebergPartitionSpec *
+GetPartitionSpecIfAlreadyExist(Oid relationId, List *partitionTransforms)
+{
+	HTAB	   *allSpecs = GetAllPartitionSpecsFromCatalog(relationId);
+
+	HASH_SEQ_STATUS currentSpecsStatus;
+
+	hash_seq_init(&currentSpecsStatus, allSpecs);
+	IcebergPartitionSpecHashEntry *currentSpec = NULL;
+
+	while ((currentSpec = hash_seq_search(&currentSpecsStatus)) != NULL)
+	{
+		/*
+		 * Partition fields in spec is already ordered by partition field id,
+		 * so we can compare them one by one with the given partition
+		 * transforms.
+		 */
+		if (PartitionTransformsEqual(currentSpec->spec, partitionTransforms))
+		{
+			IcebergPartitionSpec *foundSpec = currentSpec->spec;
+
+			hash_seq_term(&currentSpecsStatus);
+
+			return foundSpec;
+		}
+	}
+
+	return NULL;
+}
+
+/*
+* PartitionTransformsEqual compares the given partition spec
+* with the given list of partition transforms. It returns true
+* if they are equal, false otherwise.
+*/
+static bool
+PartitionTransformsEqual(IcebergPartitionSpec * spec, List *partitionTransforms)
+{
+	if (spec->fields_length != list_length(partitionTransforms))
+		return false;
+
+	for (int i = 0; i < spec->fields_length; i++)
+	{
+		IcebergPartitionSpecField *specField = &spec->fields[i];
+		IcebergPartitionTransform *transform = list_nth(partitionTransforms, i);
+
+		/*
+		 * Normally, the name check should be sufficient, as we currently do
+		 * not allow renaming any column that is used in partitioning, see
+		 * ErrorIfColumnEverUsedInIcebergPartitionSpec(). Still, let's be
+		 * defensive and also check source field ids.
+		 */
+		if (specField->source_id != transform->sourceField->id)
+			return false;
+
+		/*
+		 * This like a_bucket_10, c_year, etc., so we it includes both the
+		 * column name, the transform type and the parameters to
+		 * bucket/truncate transforms. We are essentially following what
+		 * Iceberg does here:
+		 * https://github.com/apache/iceberg/blob/8b55ac834015ce664f879ecfe1e80a941a994420/api/src/main/java/org/apache/iceberg/PartitionSpec.java#L239-L259
+		 */
+		if (strcasecmp(specField->name, transform->partitionFieldName) != 0)
+		{
+			return false;
+		}
+	}
+
+	return true;
+}
+
+
+
 /*
 * For a given relationId, this function returns the list of
 * IcebergPartitionTransform for the table with the current

pg_lake_table/tests/pytests/test_iceberg_partitions_misc.py

@@ -1,6 +1,183 @@
 from utils_pytest import *
 
 
+# show that re-setting the same partition spec
+# doesn't add a new spec, instead re-uses
+def test_re_set_partition_fields(
+    extension, s3, with_default_location, pg_conn, superuser_conn
+):
+
+    run_command(f"""CREATE SCHEMA test_re_set_partition_fields;""", pg_conn)
+
+    run_command(
+        "CREATE TABLE test_re_set_partition_fields.tbl(key int, value text) USING iceberg",
+        pg_conn,
+    )
+    pg_conn.commit()
+
+    # when there are no partitions, the default spec-id is used, and there are no corresponding entries
+    # in partition_specs
+    default_spec_id = run_query(
+        "SELECT default_spec_id FROM lake_iceberg.tables_internal WHERE table_name = 'test_re_set_partition_fields.tbl'::regclass",
+        superuser_conn,
+    )
+    assert default_spec_id[0][0] == 0
+    assert (
+        default_table_partition_spec_id(pg_conn, "test_re_set_partition_fields", "tbl")
+        == 0
+    )
+    assert (
+        len(table_partition_specs(pg_conn, "test_re_set_partition_fields", "tbl")) == 1
+    )
+
+    # now, add a spec, and make sure it is pushed
+    run_command(
+        f"ALTER TABLE test_re_set_partition_fields.tbl OPTIONS (ADD partition_by 'key')",
+        pg_conn,
+    )
+    pg_conn.commit()
+
+    default_spec_id = run_query(
+        "SELECT default_spec_id FROM lake_iceberg.tables_internal WHERE table_name = 'test_re_set_partition_fields.tbl'::regclass",
+        superuser_conn,
+    )
+    assert default_spec_id[0][0] == 1
+    assert (
+        default_table_partition_spec_id(pg_conn, "test_re_set_partition_fields", "tbl")
+        == 1
+    )
+    assert (
+        len(table_partition_specs(pg_conn, "test_re_set_partition_fields", "tbl")) == 2
+    )
+
+    # now, drop the spec, and make sure it is pushed
+    run_command(
+        f"ALTER TABLE test_re_set_partition_fields.tbl OPTIONS (DROP partition_by)",
+        pg_conn,
+    )
+    pg_conn.commit()
+
+    default_spec_id = run_query(
+        "SELECT default_spec_id FROM lake_iceberg.tables_internal WHERE table_name = 'test_re_set_partition_fields.tbl'::regclass",
+        superuser_conn,
+    )
+    assert default_spec_id[0][0] == 0
+    assert (
+        default_table_partition_spec_id(pg_conn, "test_re_set_partition_fields", "tbl")
+        == 0
+    )
+    assert (
+        len(table_partition_specs(pg_conn, "test_re_set_partition_fields", "tbl")) == 2
+    )
+
+    # now, add another partition spec
+    run_command(
+        f"ALTER TABLE test_re_set_partition_fields.tbl OPTIONS (ADD partition_by 'bucket(10, key)')",
+        pg_conn,
+    )
+    pg_conn.commit()
+
+    default_spec_id = run_query(
+        "SELECT default_spec_id FROM lake_iceberg.tables_internal WHERE table_name = 'test_re_set_partition_fields.tbl'::regclass",
+        superuser_conn,
+    )
+    assert default_spec_id[0][0] == 2
+    assert (
+        default_table_partition_spec_id(pg_conn, "test_re_set_partition_fields", "tbl")
+        == 2
+    )
+    assert (
+        len(table_partition_specs(pg_conn, "test_re_set_partition_fields", "tbl")) == 3
+    )
+
+    # setting back to existing one should not push a new id
+    # now, add a spec, and make sure it is pushed
+    run_command(
+        f"ALTER TABLE test_re_set_partition_fields.tbl OPTIONS (SET partition_by 'key')",
+        pg_conn,
+    )
+    pg_conn.commit()
+
+    default_spec_id = run_query(
+        "SELECT default_spec_id FROM lake_iceberg.tables_internal WHERE table_name = 'test_re_set_partition_fields.tbl'::regclass",
+        superuser_conn,
+    )
+    assert default_spec_id[0][0] == 1
+    assert (
+        default_table_partition_spec_id(pg_conn, "test_re_set_partition_fields", "tbl")
+        == 1
+    )
+    assert (
+        len(table_partition_specs(pg_conn, "test_re_set_partition_fields", "tbl")) == 3
+    )
+
+    # a spec with two fields swapped are considered separate specs
+    run_command(
+        f"ALTER TABLE test_re_set_partition_fields.tbl OPTIONS (SET partition_by 'truncate(100, key), bucket(50, value)')",
+        pg_conn,
+    )
+    run_command(
+        f"ALTER TABLE test_re_set_partition_fields.tbl OPTIONS (SET partition_by 'bucket(50, value), truncate(100, key)')",
+        pg_conn,
+    )
+    pg_conn.commit()
+
+    default_spec_id = run_query(
+        "SELECT default_spec_id FROM lake_iceberg.tables_internal WHERE table_name = 'test_re_set_partition_fields.tbl'::regclass",
+        superuser_conn,
+    )
+    assert default_spec_id[0][0] == 4
+    assert (
+        default_table_partition_spec_id(pg_conn, "test_re_set_partition_fields", "tbl")
+        == 4
+    )
+    assert (
+        len(table_partition_specs(pg_conn, "test_re_set_partition_fields", "tbl")) == 5
+    )
+
+    run_command(
+        f"ALTER TABLE test_re_set_partition_fields.tbl OPTIONS (SET partition_by 'truncate(100, key), bucket(50, value)')",
+        pg_conn,
+    )
+    pg_conn.commit()
+
+    default_spec_id = run_query(
+        "SELECT default_spec_id FROM lake_iceberg.tables_internal WHERE table_name = 'test_re_set_partition_fields.tbl'::regclass",
+        superuser_conn,
+    )
+    assert default_spec_id[0][0] == 3
+    assert (
+        default_table_partition_spec_id(pg_conn, "test_re_set_partition_fields", "tbl")
+        == 3
+    )
+    assert (
+        len(table_partition_specs(pg_conn, "test_re_set_partition_fields", "tbl")) == 5
+    )
+
+    # finally, drop the spec and finish the test
+    run_command(
+        f"ALTER TABLE test_re_set_partition_fields.tbl OPTIONS (DROP partition_by)",
+        pg_conn,
+    )
+    pg_conn.commit()
+
+    default_spec_id = run_query(
+        "SELECT default_spec_id FROM lake_iceberg.tables_internal WHERE table_name = 'test_re_set_partition_fields.tbl'::regclass",
+        superuser_conn,
+    )
+    assert default_spec_id[0][0] == 0
+    assert (
+        default_table_partition_spec_id(pg_conn, "test_re_set_partition_fields", "tbl")
+        == 0
+    )
+    assert (
+        len(table_partition_specs(pg_conn, "test_re_set_partition_fields", "tbl")) == 5
+    )
+
+    run_command(f"""DROP SCHEMA test_re_set_partition_fields CASCADE;""", pg_conn)
+    pg_conn.commit()
+
+
 def test_ddl_with_identity_partitions(extension, s3, with_default_location, pg_conn):
     run_command(f"""CREATE SCHEMA test_ddl_with_partitions;""", pg_conn)
 
@@ -455,3 +632,26 @@ def test_partition_by_with_collation(extension, s3, with_default_location, pg_co
 
     run_command("DROP COLLATION s_coll;", pg_conn)
     pg_conn.commit()
+
+
+def table_metadata(pg_conn, table_namespace, table_name):
+    metadata_location = run_query(
+        f"SELECT metadata_location FROM iceberg_tables WHERE table_name = '{table_name}' and table_namespace = '{table_namespace}'",
+        pg_conn,
+    )[0][0]
+
+    pg_query = f"SELECT * FROM lake_iceberg.metadata('{metadata_location}')"
+
+    metadata = run_query(pg_query, pg_conn)[0][0]
+
+    return metadata
+
+
+def table_partition_specs(pg_conn, table_namespace, table_name):
+    metadata = table_metadata(pg_conn, table_namespace, table_name)
+    return metadata["partition-specs"]
+
+
+def default_table_partition_spec_id(pg_conn, table_namespace, table_name):
+    metadata = table_metadata(pg_conn, table_namespace, table_name)
+    return metadata["default-spec-id"]