Error out at master_create_distributed_table if the table has any rows

Before this change, we do not check whether given table which already contains any data
in master_create_distributed_table command. If that table contains any data, making it
it distributed, makes that data hidden to user. With this change, we now gave error to
user if the table contains data.

src/backend/distributed/commands/create_distributed_table.c

@@ -28,6 +28,7 @@
 #include "distributed/master_metadata_utility.h"
 #include "distributed/metadata_cache.h"
 #include "distributed/pg_dist_partition.h"
+#include "executor/spi.h"
 #include "nodes/execnodes.h"
 #include "nodes/nodeFuncs.h"
 #include "nodes/pg_list.h"
@@ -48,6 +49,7 @@ static void RecordDistributedRelationDependencies(Oid distributedRelationId,
 												  Node *distributionKey);
 static Oid SupportFunctionForColumn(Var *partitionColumn, Oid accessMethodId,
 									int16 supportFunctionNumber);
+static bool LocalTableEmpty(Oid tableId);
 
 
 /* exports for SQL callable functions */
@@ -122,6 +124,17 @@ master_create_distributed_table(PG_FUNCTION_ARGS)
 								  "foreign tables.")));
 	}
 
+	/* check that the relation does not contain any rows */
+	if (!LocalTableEmpty(distributedRelationId))
+	{
+		ereport(ERROR, (errcode(ERRCODE_INVALID_TABLE_DEFINITION),
+						errmsg("cannot distribute relation \"%s\"",
+							   distributedRelationName),
+						errdetail("Relation \"%s\" contains data.",
+								  distributedRelationName),
+						errhint("Empty your table before distributing it.")));
+	}
+
 	distributionKey = BuildDistributionKeyFromColumnName(distributedRelation,
 														 distributionColumnName);
 	distributionKeyString = nodeToString(distributionKey);
@@ -377,3 +390,59 @@ SupportFunctionForColumn(Var *partitionColumn, Oid accessMethodId,
 
 	return supportFunctionOid;
 }
+
+
+/*
+ * LocalTableEmpty function checks whether given local table contains any row and
+ * returns false if there is any data. This function is only for local tables and
+ * should not be called for distributed tables.
+ */
+static bool
+LocalTableEmpty(Oid tableId)
+{
+	Oid schemaId = get_rel_namespace(tableId);
+	char *schemaName = get_namespace_name(schemaId);
+	char *tableName = get_rel_name(tableId);
+	char *tableQualifiedName = quote_qualified_identifier(schemaName, tableName);
+
+	int spiConnectionResult = 0;
+	int spiQueryResult = 0;
+	StringInfo selectExistQueryString = makeStringInfo();
+
+	HeapTuple tuple = NULL;
+	Datum hasDataDatum = 0;
+	bool localTableEmpty = false;
+	bool columnNull = false;
+	bool readOnly = true;
+
+	int rowId = 0;
+	int attributeId = 1;
+
+	AssertArg(!IsDistributedTable(tableId));
+
+	spiConnectionResult = SPI_connect();
+	if (spiConnectionResult != SPI_OK_CONNECT)
+	{
+		ereport(ERROR, (errmsg("could not connect to SPI manager")));
+	}
+
+	appendStringInfo(selectExistQueryString, SELECT_EXIST_QUERY, tableQualifiedName);
+
+	spiQueryResult = SPI_execute(selectExistQueryString->data, readOnly, 0);
+	if (spiQueryResult != SPI_OK_SELECT)
+	{
+		ereport(ERROR, (errmsg("execution was not successful \"%s\"",
+							   selectExistQueryString->data)));
+	}
+
+	/* we expect that SELECT EXISTS query will return single value in a single row */
+	Assert(SPI_processed == 1);
+
+	tuple = SPI_tuptable->vals[rowId];
+	hasDataDatum = SPI_getbinval(tuple, SPI_tuptable->tupdesc, attributeId, &columnNull);
+	localTableEmpty = !DatumGetBool(hasDataDatum);
+
+	SPI_finish();
+
+	return localTableEmpty;
+}

src/include/distributed/master_metadata_utility.h

@@ -24,6 +24,7 @@
 
 /* total number of hash tokens (2^32) */
 #define HASH_TOKEN_COUNT INT64CONST(4294967296)
+#define SELECT_EXIST_QUERY "SELECT EXISTS (SELECT 1 FROM %s)"
 
 /* In-memory representation of a typed tuple in pg_dist_shard. */
 typedef struct ShardInterval

src/test/regress/expected/multi_complex_expressions.out

@@ -390,7 +390,7 @@ ORDER BY
 	customer_keys.o_custkey DESC
 LIMIT 10 OFFSET 20;
 DEBUG:  push down of limit count: 30
-DEBUG:  building index "pg_toast_17021_index" on table "pg_toast_17021"
+DEBUG:  building index "pg_toast_17022_index" on table "pg_toast_17022"
  o_custkey | total_order_count 
 -----------+-------------------
       1466 |                 1

src/test/regress/expected/multi_create_table.out

@@ -75,6 +75,12 @@ CREATE TABLE nation (
 	n_name char(25) not null,
 	n_regionkey integer not null,
 	n_comment varchar(152));
+\COPY nation FROM STDIN WITH CSV
+SELECT master_create_distributed_table('nation', 'n_nationkey', 'append');
+ERROR:  cannot distribute relation "nation"
+DETAIL:  Relation "nation" contains data.
+HINT:  Empty your table before distributing it.
+TRUNCATE nation;
 SELECT master_create_distributed_table('nation', 'n_nationkey', 'append');
  master_create_distributed_table 
 ---------------------------------

src/test/regress/sql/multi_create_table.sql

@@ -61,6 +61,19 @@ CREATE TABLE nation (
 	n_name char(25) not null,
 	n_regionkey integer not null,
 	n_comment varchar(152));
+
+\COPY nation FROM STDIN WITH CSV
+1,'name',1,'comment_1'
+2,'name',2,'comment_2'
+3,'name',3,'comment_3'
+4,'name',4,'comment_4'
+5,'name',5,'comment_5'
+\.
+
+SELECT master_create_distributed_table('nation', 'n_nationkey', 'append');
+
+TRUNCATE nation;
+
 SELECT master_create_distributed_table('nation', 'n_nationkey', 'append');
 
 CREATE TABLE part (