diff --git a/src/backend/distributed/commands/create_distributed_table.c b/src/backend/distributed/commands/create_distributed_table.c
index 0c37c9b9f..22402b9af 100644
--- a/src/backend/distributed/commands/create_distributed_table.c
+++ b/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;
+}
diff --git a/src/include/distributed/master_metadata_utility.h b/src/include/distributed/master_metadata_utility.h
index d7610d310..dd38ae22b 100644
--- a/src/include/distributed/master_metadata_utility.h
+++ b/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
diff --git a/src/test/regress/expected/multi_complex_expressions.out b/src/test/regress/expected/multi_complex_expressions.out
index f74254f8f..0eb336111 100644
--- a/src/test/regress/expected/multi_complex_expressions.out
+++ b/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
diff --git a/src/test/regress/expected/multi_create_table.out b/src/test/regress/expected/multi_create_table.out
index d987f5f09..a41f7af07 100644
--- a/src/test/regress/expected/multi_create_table.out
+++ b/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 
 ---------------------------------
diff --git a/src/test/regress/sql/multi_create_table.sql b/src/test/regress/sql/multi_create_table.sql
index 4d9db7e37..0662c9744 100644
--- a/src/test/regress/sql/multi_create_table.sql
+++ b/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 (