You are viewing a plain text version of this content. The canonical link for it is here.

Posted to commits@hudi.apache.org by "nsivabalan (via GitHub)" <gi...@apache.org> on 2023/09/14 02:15:08 UTC

[GitHub] [hudi] nsivabalan opened a new pull request, #9712: [HUDI-6851] Fixing Spark quick start guide

nsivabalan opened a new pull request, #9712:
URL: https://github.com/apache/hudi/pull/9712

### Change Logs

Few issues to clean up in our spark quick start guide.
1. Added code snippets for keyless table in addition to primary keyed tables.
2. Simplified entire quick start to focus only on CRUD operations. For advanced operations, added references to other pages.
3. Added brief section on streaming writers and added other references.
4. Removed adhoc spark-sql commands, procedures and added references to other pages.
5. Our quick start page had references to deprecated config keys. Replaced them to use ConfigProperties.

### Impact

Improves usability for end user.

### Risk level (write none, low medium or high below)

_If medium or high, explain what verification was done to mitigate the risks._

### Documentation Update

_Describe any necessary documentation update if there is any new feature, config, or user-facing change_

- _The config description must be updated if new configs are added or the default value of the configs are changed_
- _Any new feature or user-facing change requires updating the Hudi website. Please create a Jira ticket, attach the
ticket number here and follow the [instruction](https://hudi.apache.org/contribute/developer-setup#website) to make
changes to the website._

### Contributor's checklist

- [ ] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [ ] Change Logs and Impact were stated clearly
- [ ] Adequate tests were added if applicable
- [ ] CI passed

--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] nsivabalan commented on a diff in pull request #9712: [HUDI-6851] Fixing Spark quick start guide

Posted by "nsivabalan (via GitHub)" <gi...@apache.org>.

nsivabalan commented on code in PR #9712:
URL: https://github.com/apache/hudi/pull/9712#discussion_r1326564419


##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:

Review Comment:
   For now, I have not fixed the sql_ddl and sql_dml pages yet. Can you focus on reviewing the quick start page. 



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [HUDI-6851] Fixing Spark quick start guide [hudi]

Posted by "bhasudha (via GitHub)" <gi...@apache.org>.

bhasudha merged PR #9712:
URL: https://github.com/apache/hudi/pull/9712


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] nsivabalan commented on pull request #9712: [HUDI-6851] Fixing Spark quick start guide

Posted by "nsivabalan (via GitHub)" <gi...@apache.org>.

nsivabalan commented on PR #9712:
URL: https://github.com/apache/hudi/pull/9712#issuecomment-1720229239

   Spark-Sql screenshot
   ![screencapture-localhost-3000-docs-next-quick-start-guide-2023-09-14-15_18_46](https://github.com/apache/hudi/assets/513218/cf25c6ad-f912-4c88-893b-ee1f092795d5)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] vinothchandar commented on pull request #9712: [HUDI-6851] Fixing Spark quick start guide

Posted by "vinothchandar (via GitHub)" <gi...@apache.org>.

vinothchandar commented on PR #9712:
URL: https://github.com/apache/hudi/pull/9712#issuecomment-1729577986

   Open items 
   
   - [ ] Consistent table names and use of data across DataFrame vs SQL (use either trips across all or use synthetic across all)


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] bhasudha commented on a diff in pull request #9712: [HUDI-6851] Fixing Spark quick start guide

Posted by "bhasudha (via GitHub)" <gi...@apache.org>.

bhasudha commented on code in PR #9712:
URL: https://github.com/apache/hudi/pull/9712#discussion_r1330703184


##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
+
+### Table Type
+Here is an example of creating a COW table.
+
+```sql
+-- create a non-primary key (or key less) table
+create table if not exists hudi_table2(
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow'
+);
+```
+
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 
+the table and hudi will treat them as key less table. Users can still perform Merge Into, updates and deletes based on any random data column. 
+
+### Primary Key
+Here is an example of creating COW table with a primary key 'id'. For mutable datasets, it is recommended to set appropriate primary key. 
+
+```sql
+-- create a managed cow table
+create table if not exists hudi_table0 (
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+);
+```
+
+### PreCombineField
+Here is an example of creating an MOR external table. The **preCombineField** option
+is used to specify the preCombine field for merge. Generally 'event time' or some other similar column will be used for 
+ordering purpose. Hudi will be able to handle out of order data using the precombine field value.
+
+```sql
+-- create an external mor table
+create table if not exists hudi_table1 (
+  id int, 
+  name string, 
+  price double,
+  ts bigint
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id,name',
+  preCombineField = 'ts' 
+);
+```
+
+### Partitioned Table
+Here is an example of creating a COW partitioned key less table.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+ ) 
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table
+Here is an example of creating a COW un-partitioned table.
+
+```sql
+-- create a cow table, with primaryKey 'uuid' and unpartitioned. 
+create table hudi_cow_nonpcf_tbl (
+  uuid int,
+  name string,
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'uuid'
+);
+```
+
+Here is an example of creating a MOR un-partitioned table.
+
+```sql
+-- create a mor non-partitioned table with preCombineField provided
+create table hudi_mor_tbl (
+    id int,
+    name string,
+    price double,
+    ts bigint
+) using hudi
+tblproperties (
+    type = 'mor',
+    primaryKey = 'id',
+    preCombineField = 'ts'
+);
+```
+
+### Create Table for an External Hudi Table
+You can create an External table using the `location` statement. If an external location is not specified it is considered a managed table. 
+You can read more about external vs managed tables [here](https://sparkbyexamples.com/apache-hive/difference-between-hive-internal-tables-and-external-tables/).
+An external table is useful if you need to read/write to/from a pre-existing hudi table.
+
+```sql
+ create table h_p1 using hudi
+ location '/path/to/hudi';
+```
+
+:::tip
+You don't need to specify schema and any properties except the partitioned columns if existed. Hudi can automatically recognize the schema and configurations.
+:::
+
+### Create Table AS SELECT
+
+Hudi supports CTAS(Create table as select) on spark sql. <br/>
+**Note:** For better performance to load data to hudi table, CTAS uses **bulk insert** as the write operation.
+
+**Example CTAS command to create a non-partitioned COW key less table.**
+
+```sql 
+create table h3 using hudi
+tblproperties (type = 'cow')
+as
+select 1 as id, 'a1' as name, 10 as price;
+```
+
+**Example CTAS command to create a partitioned, primary keyed COW table.**
+
+```sql
+create table h2 using hudi
+tblproperties (type = 'cow', primaryKey = 'id')
+partitioned by (dt)
+as
+select 1 as id, 'a1' as name, 10 as price, 1000 as dt;
+```
+
+**Example CTAS command to load data from another table.**
+
+```sql
+# create managed parquet table 
+create table parquet_mngd using parquet location 'file:///tmp/parquet_dataset/*.parquet';
+
+# CTAS by loading data into hudi table
+create table hudi_tbl using hudi location 'file:/tmp/hudi/hudi_tbl/' tblproperties ( 
+  type = 'cow', 
+  primaryKey = 'id', 
+  preCombineField = 'ts' 
+ ) 
+partitioned by (datestr) as select * from parquet_mngd;
+```
+
+### Set hoodie config options
+You can also set the config with table options when creating table which will work for
+the table scope only and override the config set by the SET command.
+```sql
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  ${hoodie.config.key1} = '${hoodie.config.value2}',
+  ${hoodie.config.key2} = '${hoodie.config.value2}',
+  ....
+);
+
+e.g.
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  hoodie.cleaner.fileversions.retained = '20',
+  hoodie.keep.max.commits = '20'
+);
+```
+
+## Spark Alter Table
+### Syntax
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+```
+
+:::note
+`ALTER TABLE ... RENAME TO ...` is not supported when using AWS Glue Data Catalog as hive metastore as Glue itself does 
+not support table renames.
+:::
+
+### Examples
+```sql
+alter table h0 rename to h0_1;
+
+alter table h0_1 add columns(ext0 string);
+
+alter table h0_1 change column id id bigint;
+```
+
+Schema evolution can be achieved via `ALTER TABLE` commands. Below shows some basic examples.
+
+:::note
+For more detailed examples, please prefer to [schema evolution](/docs/schema_evolution)
+:::
+
+**Syntax**
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+
+-- Alter table properties
+ALTER TABLE tableIdentifier SET TBLPROPERTIES (key = 'value')
+```
+
+**Examples**
+```sql
+--rename to:
+ALTER TABLE hudi_cow_nonpcf_tbl RENAME TO hudi_cow_nonpcf_tbl2;
+
+--add column:
+ALTER TABLE hudi_cow_nonpcf_tbl2 add columns(remark string);
+
+--change column:
+ALTER TABLE hudi_cow_nonpcf_tbl2 change column uuid uuid bigint;
+
+--set properties;
+alter table hudi_cow_nonpcf_tbl2 set tblproperties (hoodie.keep.max.commits = '10');
+```
+
+### Alter hoodie config options
+You can also alter the write config for a table by the **ALTER SERDEPROPERTIES**
+
+Example:
+```sql
+ alter table h3 set serdeproperties (hoodie.keep.max.commits = '10') 
+```
+
+### Use set command
+You can use the **set** command to set any custom hudi's config, which will work for the
+whole spark session scope.
+```sql
+set hoodie.insert.shuffle.parallelism = 100;
+set hoodie.upsert.shuffle.parallelism = 100;
+set hoodie.delete.shuffle.parallelism = 100;
+```
+
+### Show and Drop partitions
+
+**Syntax**
+
+```sql
+-- Show Partitions
+SHOW PARTITIONS tableIdentifier
+
+-- Drop Partition
+ALTER TABLE tableIdentifier DROP PARTITION ( partition_col_name = partition_col_val [ , ... ] )
+```
+
+**Examples**
+```sql
+--show partition:
+show partitions hudi_cow_pt_tbl;
+
+--drop partition：
+alter table hudi_cow_pt_tbl drop partition (dt='2021-12-09', hh='10');
+```
+
+:::note
+Currently,  the result of `show partitions` is based on the filesystem table path. It's not precise when delete the whole partition data or drop certain partition directly.
+:::
+
+## Flink
+
+### Create Catalog
+
+The catalog helps to manage the SQL tables, the table can be shared among CLI sessions if the catalog persists the table DDLs.
+For `hms` mode, the catalog also supplements the hive syncing options.
+
+HMS mode catalog SQL demo:
+```sql
+CREATE CATALOG hoodie_catalog
+  WITH (
+    'type'='hudi',
+    'catalog.path' = '${catalog default root path}',
+    'hive.conf.dir' = '${directory where hive-site.xml is located}',
+    'mode'='hms' -- supports 'dfs' mode that uses the DFS backend for table DDLs persistence
+  );
+```
+
+#### Options
+|  Option Name  | Required | Default | Remarks |
+|  -----------  | -------  | ------- | ------- |
+| `catalog.path` | true | -- | Default root path for the catalog, the path is used to infer the table path automatically, the default table path: `${catalog.path}/${db_name}/${table_name}` |
+| `default-database` | false | default | default database name |
+| `hive.conf.dir` | false | -- | The directory where hive-site.xml is located, only valid in `hms` mode |
+| `mode` | false | dfs | Supports `hms` mode that uses HMS to persist the table options |

Review Comment:
   @danny0405  does all possible values - 'hms', 'glue', 'hiveql', 'jdbc' work here?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] nsivabalan commented on a diff in pull request #9712: [HUDI-6851] Fixing Spark quick start guide

Posted by "nsivabalan (via GitHub)" <gi...@apache.org>.

nsivabalan commented on code in PR #9712:
URL: https://github.com/apache/hudi/pull/9712#discussion_r1328273936


##########
website/docs/quick-start-guide.md:
##########
@@ -73,6 +65,13 @@ spark-shell \
   --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension'
 ```
 ```shell
+# Spark 3.0

Review Comment:
   We are adding back support for spark3.0 and hence showing launch command for spark3.0. 



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] bhasudha commented on a diff in pull request #9712: [HUDI-6851] Fixing Spark quick start guide

Posted by "bhasudha (via GitHub)" <gi...@apache.org>.

bhasudha commented on code in PR #9712:
URL: https://github.com/apache/hudi/pull/9712#discussion_r1330709219


##########
website/docs/quick-start-guide.md:
##########
@@ -7,10 +7,8 @@ last_modified_at: 2023-08-23T21:14:52+09:00
 import Tabs from '@theme/Tabs';
 import TabItem from '@theme/TabItem';
 
-This guide provides a quick peek at Hudi's capabilities using spark-shell. Using Spark datasources, we will walk through
-code snippets that allows you to insert and update a Hudi table of default table type:
-[Copy on Write](/docs/table_types#copy-on-write-table). After each write operation we will also show how to read the
-data both snapshot and incrementally.
+This guide provides a quick peek at Hudi's capabilities using spark. Using Spark datasources, pyspark and Spark SQL, 

Review Comment:
   fixed



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] nsivabalan commented on a diff in pull request #9712: [HUDI-6851] Fixing Spark quick start guide

Posted by "nsivabalan (via GitHub)" <gi...@apache.org>.

nsivabalan commented on code in PR #9712:
URL: https://github.com/apache/hudi/pull/9712#discussion_r1326566706


##########
website/docs/quick-start-guide.md:
##########
@@ -778,18 +727,20 @@ val updates = convertToStringList(dataGen.generateUpdates(10))
 val df = spark.read.json(spark.sparkContext.parallelize(updates, 2))
 df.write.format("hudi").
   options(getQuickstartWriteConfigs).
-  option(PRECOMBINE_FIELD_OPT_KEY, "ts").
-  option(RECORDKEY_FIELD_OPT_KEY, "uuid").
-  option(PARTITIONPATH_FIELD_OPT_KEY, "partitionpath").
+  option(PRECOMBINE_FIELD_NAME.key(), "ts").
+  option(RECORDKEY_FIELD_NAME.key(), "uuid").
+  option(PARTITIONPATH_FIELD_NAME.key(), "partitionpath").
   option(TABLE_NAME, tableName).
   mode(Append).
   save(basePath)
 ```
 :::note
-Notice that the save mode is now `Append`. In general, always use append mode unless you are trying to create the table for the first time.
-[Querying](#query-data) the data again will now show updated trips. Each write operation generates a new [commit](/docs/concepts)
+- Notice that the save mode is now `Append`. In general, always use append mode unless you are trying to create the table for the first time.
+[Querying](#query-data) the data again will now show updated trips. Each write operation generates a new [commit](/docs/next/concepts)
 denoted by the timestamp. Look for changes in `_hoodie_commit_time`, `rider`, `driver` fields for the same `_hoodie_record_key`s in previous commit.
+- Updates to a Keyless table via spark datasource is not supported. Updates for keyless tables are supported with Spark-SQL MergeInto and UPDATE commands.

Review Comment:
   yes.



##########
website/docs/quick-start-guide.md:
##########
@@ -866,6 +817,12 @@ when not matched then
 
 ```
 
+:::note
+For a Primary keyed Hudi table, the join condition in MIT is expected to contain the primary keys of the table. 

Review Comment:
   Merge Into



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] nsivabalan commented on pull request #9712: [HUDI-6851] Fixing Spark quick start guide

Posted by "nsivabalan (via GitHub)" <gi...@apache.org>.

nsivabalan commented on PR #9712:
URL: https://github.com/apache/hudi/pull/9712#issuecomment-1720227705

   Spark-sql screenshot
   ![screencapture-localhost-3000-docs-next-quick-start-guide-2023-09-14-15_04_03 (1)](https://github.com/apache/hudi/assets/513218/6ee4b578-6416-41e4-a33a-85ee113ec931)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] vinothchandar commented on a diff in pull request #9712: [HUDI-6851] Fixing Spark quick start guide

Posted by "vinothchandar (via GitHub)" <gi...@apache.org>.

vinothchandar commented on code in PR #9712:
URL: https://github.com/apache/hudi/pull/9712#discussion_r1327983115


##########
website/docs/quick-start-guide.md:
##########
@@ -7,10 +7,8 @@ last_modified_at: 2023-08-23T21:14:52+09:00
 import Tabs from '@theme/Tabs';
 import TabItem from '@theme/TabItem';
 
-This guide provides a quick peek at Hudi's capabilities using spark-shell. Using Spark datasources, we will walk through
-code snippets that allows you to insert and update a Hudi table of default table type:
-[Copy on Write](/docs/table_types#copy-on-write-table). After each write operation we will also show how to read the
-data both snapshot and incrementally.
+This guide provides a quick peek at Hudi's capabilities using spark. Using Spark datasources, pyspark and Spark SQL, 

Review Comment:
   "datasources" ?
   



##########
website/docs/quick-start-guide.md:
##########
@@ -73,6 +65,13 @@ spark-shell \
   --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension'
 ```
 ```shell
+# Spark 3.0

Review Comment:
   why is this Spark 3.0 , when the default build is Spark 3.3?



##########
website/docs/quick-start-guide.md:
##########
@@ -216,9 +237,33 @@ can generate sample inserts and updates based on the the sample trip schema [her
 
 ## Create Table
 
+Before we go further, few terminologies to familiarize: 
+
+- **Table types**

Review Comment:
   We don't need anything like this here. lets just assume a table type and go from there. lets pick MoR vs CoW based what can show case breadth easily, same for partitioned vs non-partitioned. 



##########
website/docs/quick-start-guide.md:
##########
@@ -216,9 +237,33 @@ can generate sample inserts and updates based on the the sample trip schema [her
 
 ## Create Table
 
+Before we go further, let us go over few terminologies: 
+
+- **Table types**
+
+  Hudi supports two different table types, namely Copy-On-Write (COW) and Merge-On-Read (MOR). Users can choose either 
+of these table types depending on their workload and SLA requirements. You can read more about different 
+  table types [here](/docs/next/table_types/).
+
+- **Partitioned & Non-Partitioned tables**
+
+  Users can create a partitioned table or a non-partitioned table with Apache Hudi. Partitioning can help with 
+  reducing query run times. For quick start purpose, we will go with partitioned table. 
+
+- **Primary key and Hudi table**
+
+  Optionally users can choose to create a Primary keyed table. When primary key is set for a given table,
+  Hudi ensures uniqueness during updates and deletes. Each record is uniquely identified by the primary key configuration.
+  If primary key is not set, Hudi treats it as key less table and every record ingested is treated as a new record even
+  if contents match. Such keyless tables are supported from Hudi 0.14.0.

Review Comment:
   and this line talks about 0.14.0 vs 0.13.0?



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,408 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."

Review Comment:
   this page is specific to Spark right? it may be confusing to readers if we don't qualify all that. 



##########
website/docs/quick-start-guide.md:
##########
@@ -73,6 +65,13 @@ spark-shell \
   --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension'
 ```
 ```shell
+# Spark 3.0
+spark-shell \
+  --packages org.apache.hudi:hudi-spark3.0-bundle_2.12:0.13.0 \

Review Comment:
   why does this have 0.13.0?



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,408 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+# Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 

Review Comment:
   what does scala or python mean? Can we be specific around the Spark apis, for e.g  "DataSource APIs (Batch or Streaming) via Scala or Python"



##########
website/docs/quick-start-guide.md:
##########
@@ -216,9 +237,33 @@ can generate sample inserts and updates based on the the sample trip schema [her
 
 ## Create Table
 
+Before we go further, let us go over few terminologies: 
+
+- **Table types**
+
+  Hudi supports two different table types, namely Copy-On-Write (COW) and Merge-On-Read (MOR). Users can choose either 
+of these table types depending on their workload and SLA requirements. You can read more about different 
+  table types [here](/docs/next/table_types/).
+
+- **Partitioned & Non-Partitioned tables**
+
+  Users can create a partitioned table or a non-partitioned table with Apache Hudi. Partitioning can help with 
+  reducing query run times. For quick start purpose, we will go with partitioned table. 
+
+- **Primary key and Hudi table**
+
+  Optionally users can choose to create a Primary keyed table. When primary key is set for a given table,

Review Comment:
   lets not create more artificial distinctions for the tables here. Hudi tables always have a key, its a matter of whether its user defined or system generated.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] bhasudha commented on a diff in pull request #9712: [HUDI-6851] Fixing Spark quick start guide

Posted by "bhasudha (via GitHub)" <gi...@apache.org>.

bhasudha commented on code in PR #9712:
URL: https://github.com/apache/hudi/pull/9712#discussion_r1327198462


##########
website/docs/quick-start-guide.md:
##########
@@ -778,18 +724,20 @@ val updates = convertToStringList(dataGen.generateUpdates(10))
 val df = spark.read.json(spark.sparkContext.parallelize(updates, 2))

Review Comment:
   Understand there is a note below the code snippet, on spark datasource not supporting updates/deletes on keyless table. It might be good to call out in a comment line at the start of the snippet, that, this update is on primary keyed table only as a description line. 



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] bhasudha commented on a diff in pull request #9712: [HUDI-6851] Fixing Spark quick start guide

Posted by "bhasudha (via GitHub)" <gi...@apache.org>.

bhasudha commented on code in PR #9712:
URL: https://github.com/apache/hudi/pull/9712#discussion_r1326551249


##########
website/docs/quick-start-guide.md:
##########
@@ -216,9 +237,33 @@ can generate sample inserts and updates based on the the sample trip schema [her
 
 ## Create Table
 
+Before we go further, few terminologies to familiarize: 
+
+- **Table types**

Review Comment:
   @nsivabalan   Since these are common background knowledge and nothing specific to Spark SQL, moving these concepts to even above the Setup step would be good. May be we can add a section called Background or Introduction. What do you think ?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] nsivabalan commented on pull request #9712: [HUDI-6851] Fixing Spark quick start guide

Posted by "nsivabalan (via GitHub)" <gi...@apache.org>.

nsivabalan commented on PR #9712:
URL: https://github.com/apache/hudi/pull/9712#issuecomment-1720227155

   Spark-scala sreenshot 
   
   ![screencapture-localhost-3000-docs-next-quick-start-guide-2023-09-14-15_04_03](https://github.com/apache/hudi/assets/513218/68cc7799-bd3e-46e5-affe-7d7a4edc0754)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] nsivabalan commented on a diff in pull request #9712: [HUDI-6851] Fixing Spark quick start guide

Posted by "nsivabalan (via GitHub)" <gi...@apache.org>.

nsivabalan commented on code in PR #9712:
URL: https://github.com/apache/hudi/pull/9712#discussion_r1326568180


##########
website/docs/quick-start-guide.md:
##########
@@ -1538,174 +998,205 @@ values={[
 <TabItem value="scala">
 
 ```scala
-// spark-shell
-spark.
-  read.format("hudi").
-  load(basePath).
-  select("uuid","partitionpath").
-  sort("partitionpath","uuid").
-  show(100, false)
+spark.read.
+  format("hudi").
+  option("as.of.instant", "20210728141108100").
+  load(basePath)
 
-val inserts = convertToStringList(dataGen.generateInserts(10))
-val df = spark.
-  read.json(spark.sparkContext.parallelize(inserts, 2)).
-  filter("partitionpath = 'americas/united_states/san_francisco'")
-df.write.format("hudi").
-  options(getQuickstartWriteConfigs).
-  option(OPERATION.key(),"insert_overwrite").
-  option(PRECOMBINE_FIELD.key(), "ts").
-  option(RECORDKEY_FIELD.key(), "uuid").
-  option(PARTITIONPATH_FIELD.key(), "partitionpath").
-  option(TBL_NAME.key(), tableName).
-  mode(Append).
-  save(basePath)
+spark.read.
+  format("hudi").
+  option("as.of.instant", "2021-07-28 14:11:08.200").
+  load(basePath)
+
+// It is equal to "as.of.instant = 2021-07-28 00:00:00"
+spark.read.
+  format("hudi").
+  option("as.of.instant", "2021-07-28").
+  load(basePath)
 
-// Should have different keys now for San Francisco alone, from query before.
-spark.
-  read.format("hudi").
-  load(basePath).
-  select("uuid","partitionpath").
-  sort("partitionpath","uuid").
-  show(100, false)
 ```
-</TabItem>
 
+</TabItem>
 <TabItem value="python">
 
 ```python
-# pyspark
-spark.read.format("hudi"). \
-    load(basePath). \
-    select(["uuid", "partitionpath"]). \
-    sort(["partitionpath", "uuid"]). \
-    show(n=100, truncate=False)
-    
-inserts = sc._jvm.org.apache.hudi.QuickstartUtils.convertToStringList(dataGen.generateInserts(10)) 
-df = spark.read.json(spark.sparkContext.parallelize(inserts, 2)). \
-    filter("partitionpath = 'americas/united_states/san_francisco'")
-hudi_insert_overwrite_options = {
-    'hoodie.table.name': tableName,
-    'hoodie.datasource.write.recordkey.field': 'uuid',
-    'hoodie.datasource.write.partitionpath.field': 'partitionpath',
-    'hoodie.datasource.write.table.name': tableName,
-    'hoodie.datasource.write.operation': 'insert_overwrite',
-    'hoodie.datasource.write.precombine.field': 'ts',
-    'hoodie.upsert.shuffle.parallelism': 2,
-    'hoodie.insert.shuffle.parallelism': 2
-}
-df.write.format("hudi").options(**hudi_insert_overwrite_options).mode("append").save(basePath)
-spark.read.format("hudi"). \
-    load(basePath). \
-    select(["uuid", "partitionpath"]). \
-    sort(["partitionpath", "uuid"]). \
-    show(n=100, truncate=False)
+#pyspark
+spark.read. \
+  format("hudi"). \
+  option("as.of.instant", "20210728141108"). \
+  load(basePath)
+
+spark.read. \
+  format("hudi"). \
+  option("as.of.instant", "2021-07-28 14:11:08.000"). \
+  load(basePath)
+
+# It is equal to "as.of.instant = 2021-07-28 00:00:00"
+spark.read. \
+  format("hudi"). \
+  option("as.of.instant", "2021-07-28"). \
+  load(basePath)
 ```
-</TabItem>
 
+</TabItem>
 <TabItem value="sparksql">
 
-`insert overwrite` a partitioned table use the `INSERT_OVERWRITE` type of write operation, while a non-partitioned table to `INSERT_OVERWRITE_TABLE`.
+:::note
+Requires Spark 3.2+
+:::
 
 ```sql
--- insert overwrite non-partitioned table
-insert overwrite hudi_mor_tbl select 99, 'a99', 20.0, 900;
-insert overwrite hudi_cow_nonpcf_tbl select 99, 'a99', 20.0;
 
--- insert overwrite partitioned table with dynamic partition
-insert overwrite table hudi_cow_pt_tbl select 10, 'a10', 1100, '2021-12-09', '10';
-
--- insert overwrite partitioned table with static partition
-insert overwrite hudi_cow_pt_tbl partition(dt = '2021-12-09', hh='12') select 13, 'a13', 1100;
+-- time travel based on commit time, for eg: `20220307091628793`
+select * from hudi_cow_pt_tbl timestamp as of '20220307091628793' where id = 1;
+-- time travel based on different timestamp formats
+select * from hudi_cow_pt_tbl timestamp as of '2022-03-07 09:16:28.100' where id = 1;
+select * from hudi_cow_pt_tbl timestamp as of '2022-03-08' where id = 1;
 ```
+
 </TabItem>
 
 </Tabs
 >
 
-## More Spark SQL Commands
 
-### Alter Table
+### Incremental query
 
-Schema evolution can be achieved via `ALTER TABLE` commands. Below shows some basic examples.
+Hudi also provides capability to obtain a stream of records that changed since given commit timestamp. 
+This can be achieved using Hudi's incremental querying and providing a begin time from which changes need to be streamed. 
+We do not need to specify endTime, if we want all changes after the given commit (as is the common case). 
 
-:::note
-For more detailed examples, please prefer to [schema evolution](schema_evolution)
-:::
+<Tabs
+groupId="programming-language"
+defaultValue="scala"
+values={[
+{ label: 'Scala', value: 'scala', },
+{ label: 'Python', value: 'python', },
+{ label: 'Spark SQL', value: 'sparksql', }
+]}
+>
 
-**Syntax**
-```sql
--- Alter table name
-ALTER TABLE oldTableName RENAME TO newTableName
+<TabItem value="scala">
+
+```scala
+// spark-shell
+// reload data
+spark.
+  read.
+  format("hudi").
+  load(basePath).
+  createOrReplaceTempView("hudi_trips_snapshot")
 
--- Alter table add columns
-ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+val commits = spark.sql("select distinct(_hoodie_commit_time) as commitTime from  hudi_trips_snapshot order by commitTime").map(k => k.getString(0)).take(50)
+val beginTime = commits(commits.length - 2) // commit time we are interested in
 
--- Alter table column type
-ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+// incrementally query data
+val tripsIncrementalDF = spark.read.format("hudi").
+  option(QUERY_TYPE.key(), QUERY_TYPE_INCREMENTAL_OPT_VAL).
+  option(BEGIN_INSTANTTIME.key(), beginTime).
+  load(basePath)
+tripsIncrementalDF.createOrReplaceTempView("hudi_trips_incremental")
 
--- Alter table properties
-ALTER TABLE tableIdentifier SET TBLPROPERTIES (key = 'value')
+spark.sql("select `_hoodie_commit_time`, fare, begin_lon, begin_lat, ts from  hudi_trips_incremental where fare > 20.0").show()
 ```
-**Examples**
-```sql
---rename to:
-ALTER TABLE hudi_cow_nonpcf_tbl RENAME TO hudi_cow_nonpcf_tbl2;
 
---add column:
-ALTER TABLE hudi_cow_nonpcf_tbl2 add columns(remark string);
+</TabItem>
+<TabItem value="python">
+
+```python
+# pyspark
+# reload data
+spark. \
+  read. \
+  format("hudi"). \
+  load(basePath). \
+  createOrReplaceTempView("hudi_trips_snapshot")
+
+commits = list(map(lambda row: row[0], spark.sql("select distinct(_hoodie_commit_time) as commitTime from  hudi_trips_snapshot order by commitTime").limit(50).collect()))
+beginTime = commits[len(commits) - 2] # commit time we are interested in
+
+# incrementally query data
+incremental_read_options = {
+  'hoodie.datasource.query.type': 'incremental',
+  'hoodie.datasource.read.begin.instanttime': beginTime,
+}
 
---change column:
-ALTER TABLE hudi_cow_nonpcf_tbl2 change column uuid uuid bigint;
+tripsIncrementalDF = spark.read.format("hudi"). \
+  options(**incremental_read_options). \
+  load(basePath)
+tripsIncrementalDF.createOrReplaceTempView("hudi_trips_incremental")
 
---set properties;
-alter table hudi_cow_nonpcf_tbl2 set tblproperties (hoodie.keep.max.commits = '10');
+spark.sql("select `_hoodie_commit_time`, fare, begin_lon, begin_lat, ts from  hudi_trips_incremental where fare > 20.0").show()
 ```
 
-### Partition SQL Command
+</TabItem>
 
-**Syntax**
+<TabItem value="sparksql">
 
 ```sql
--- Add Partition
-ALTER TABLE tableIdentifier ADD PARTITION ( partition_col_name = partition_col_val [ , ... ] )
+-- syntax
+hudi_table_changes(table or path, queryType, beginTime [, endTime]);  
+-- table or path: table identifier, example: db.tableName, tableName, 
+--                or path for of your table, example: path/to/hudiTable  
+--                in this case table does not need to exist in the metastore,
+-- queryType: incremental query mode, example: latest_state, cdc  
+--            (for cdc query, first enable cdc for your table by setting cdc.enabled=true),
+-- beginTime: instantTime to begin query from, example: earliest, 202305150000, 
+-- endTime: optional instantTime to end query at, example: 202305160000, 
 
--- Drop Partition
-ALTER TABLE tableIdentifier DROP PARTITION ( partition_col_name = partition_col_val [ , ... ] )
+-- incrementally query data by table name
+-- start from earliest available commit, end at latest available commit.  
+select * from hudi_table_changes('db.table', 'latest_state', 'earliest');
 
--- Show Partitions
-SHOW PARTITIONS tableIdentifier
-```
-**Examples**
-```sql
---show partition:
-show partitions hudi_cow_pt_tbl;
+-- start from earliest, end at 202305160000.  
+select * from hudi_table_changes('table', 'latest_state', 'earliest', '202305160000');  
+
+-- start from 202305150000, end at 202305160000.
+select * from hudi_table_changes('table', 'latest_state', '202305150000', '202305160000');
+
+-- incrementally query data by path
+-- start from earliest available commit, end at latest available commit.
+select * from hudi_table_changes('path/to/table', 'cdc', 'earliest');
+
+-- start from earliest, end at 202305160000.
+select * from hudi_table_changes('path/to/table', 'cdc', 'earliest', '202305160000');
+
+-- start from 202305150000, end at 202305160000.
+select * from hudi_table_changes('path/to/table', 'cdc', '202305150000', '202305160000');
 
---drop partition：
-alter table hudi_cow_pt_tbl drop partition (dt='2021-12-09', hh='10');
 ```
-:::note
-Currently,  the result of `show partitions` is based on the filesystem table path. It's not precise when delete the whole partition data or drop certain partition directly.
 
+</TabItem>
+
+</Tabs
+>
+
+:::info
+This will give all changes that happened after the beginTime commit with the filter of fare > 20.0. The unique thing about this
+feature is that it now lets you author streaming pipelines on batch data.
 :::
 
-### Procedures
+## Streaming writers
+### Hudi Streamer
+Hudi provides a ingestion tool to assist with ingesting data into hudi from various difference sources in a streaming manner. 
+This has lot of niceties like auto checkpointing, schema enforcement via schema provider, transformation support and so on.
+Please refer to [here](/docs/next/hoodie_deltastreamer#hudi-streamer) for more info. 
 
-**Syntax**
-```sql
---Call procedure by positional arguments
-CALL system.procedure_name(arg_1, arg_2, ... arg_n)
+### Structured Streaming
 
---Call procedure by named arguments
-CALL system.procedure_name(arg_name_2 => arg_2, arg_name_1 => arg_1, ... arg_name_n => arg_n)
-```
-**Examples**
-```sql
---show commit's info
-call show_commits(table => 'test_hudi_table', limit => 10);
-```
+Hudi supports Spark Structured Streaming reads and writes as well. Please refer to [here](/docs/next/toBeFixed) for more info.

Review Comment:
   yes, once we land this https://github.com/apache/hudi/pull/9701, will update it here



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] nsivabalan commented on a diff in pull request #9712: [HUDI-6851] Fixing Spark quick start guide

Posted by "nsivabalan (via GitHub)" <gi...@apache.org>.

nsivabalan commented on code in PR #9712:
URL: https://github.com/apache/hudi/pull/9712#discussion_r1326842404


##########
website/docs/quick-start-guide.md:
##########
@@ -216,9 +237,33 @@ can generate sample inserts and updates based on the the sample trip schema [her
 
 ## Create Table
 
+Before we go further, few terminologies to familiarize: 
+
+- **Table types**

Review Comment:
   lets wait to hear from vinoth as well.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] bhasudha commented on pull request #9712: [HUDI-6851] Fixing Spark quick start guide

Posted by "bhasudha (via GitHub)" <gi...@apache.org>.

bhasudha commented on PR #9712:
URL: https://github.com/apache/hudi/pull/9712#issuecomment-1726525617

   New quick start page looks like this:
   
   ![Screenshot 2023-09-19 at 2 19 12 PM (2)](https://github.com/apache/hudi/assets/2179254/da9714a7-bc52-4982-8761-db41f16315f1)
   ![Screenshot 2023-09-19 at 2 19 28 PM (2)](https://github.com/apache/hudi/assets/2179254/31d716ed-efb5-49bc-ac65-bae96edf8c4f)
   ![Screenshot 2023-09-19 at 2 19 34 PM (2)](https://github.com/apache/hudi/assets/2179254/be2c5b8f-b27b-4853-a52d-aa7cc9146054)
   ![Screenshot 2023-09-19 at 2 19 39 PM (2)](https://github.com/apache/hudi/assets/2179254/26a1c18d-acfa-43dd-ab57-b913751ae734)
   ![Screenshot 2023-09-19 at 2 19 45 PM (2)](https://github.com/apache/hudi/assets/2179254/0e919f8a-f608-424a-852b-3f93a53b968d)
   ![Screenshot 2023-09-19 at 2 19 49 PM (2)](https://github.com/apache/hudi/assets/2179254/350b1ccf-e1c8-4b51-bdf5-0ba59f25faf7)
   ![Screenshot 2023-09-19 at 2 19 52 PM (2)](https://github.com/apache/hudi/assets/2179254/052e2b92-ac91-4e8c-966a-e4267e42dcbf)
   ![Screenshot 2023-09-19 at 2 19 57 PM (2)](https://github.com/apache/hudi/assets/2179254/b92111c4-8eb1-44c5-a610-3ccf6a213df2)
   ![Screenshot 2023-09-19 at 2 20 01 PM (2)](https://github.com/apache/hudi/assets/2179254/2633fa5f-34b6-493e-97b3-5df654511eb1)
   ![Screenshot 2023-09-19 at 2 20 05 PM (2)](https://github.com/apache/hudi/assets/2179254/f6ffb052-256f-4980-86c7-ebbcb58dc871)


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] nsivabalan commented on a diff in pull request #9712: [HUDI-6851] Fixing Spark quick start guide

Posted by "nsivabalan (via GitHub)" <gi...@apache.org>.

nsivabalan commented on code in PR #9712:
URL: https://github.com/apache/hudi/pull/9712#discussion_r1327856715


##########
website/docs/sql_dml.md:
##########
@@ -0,0 +1,189 @@
+---
+title: SQL DML
+summary: "In this page, we go will cover details on how to modify data with Hudi tables"
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+
+# Spark DML Operations
+
+The following are SparkSQL DML actions available:
+
+Insert Into
+Merge Into
+Update
+Delete
+
+You can ingest data, modify or delete data from a given hudi table using these operations. Lets go over them one by one. 
+
+Please refer to [SQL DDL](/docs/next/sql_ddl) for creating Hudi tables. This page goes over ingesting and modifying data once the
+table is created. 
+
+## Insert Into

Review Comment:
   Note to reviewer:
   for now, copied over contents from our quick start page. lets land the first version. we can enhance w/ more examples once we have the first cut landed. 



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] pratyakshsharma commented on a diff in pull request #9712: [HUDI-6851] Fixing Spark quick start guide

Posted by "pratyakshsharma (via GitHub)" <gi...@apache.org>.

pratyakshsharma commented on code in PR #9712:
URL: https://github.com/apache/hudi/pull/9712#discussion_r1326039615


##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:

Review Comment:
   nit: The following DDL actions are available via SparkSQL?



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .

Review Comment:
   nit: "Set hudi config options" section. 
   
   Let us add a link to this section here.



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|

Review Comment:
   nit: two versions -> multiple versions



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
+
+### Table Type
+Here is an example of creating a COW table.
+
+```sql
+-- create a non-primary key (or key less) table
+create table if not exists hudi_table2(
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow'
+);
+```
+
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 
+the table and hudi will treat them as key less table. Users can still perform Merge Into, updates and deletes based on any random data column. 
+
+### Primary Key
+Here is an example of creating COW table with a primary key 'id'. For mutable datasets, it is recommended to set appropriate primary key. 
+
+```sql
+-- create a managed cow table
+create table if not exists hudi_table0 (
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+);
+```
+
+### PreCombineField
+Here is an example of creating an MOR external table. The **preCombineField** option
+is used to specify the preCombine field for merge. Generally 'event time' or some other similar column will be used for 
+ordering purpose. Hudi will be able to handle out of order data using the precombine field value.
+
+```sql
+-- create an external mor table
+create table if not exists hudi_table1 (
+  id int, 
+  name string, 
+  price double,
+  ts bigint
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id,name',
+  preCombineField = 'ts' 
+);
+```
+
+### Partitioned Table
+Here is an example of creating a COW partitioned key less table.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+ ) 
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table

Review Comment:
   nit: Let us change it to Non-Partitioned Table to have uniform terminologies throughout.



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
+
+### Table Type
+Here is an example of creating a COW table.
+
+```sql
+-- create a non-primary key (or key less) table
+create table if not exists hudi_table2(
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow'
+);
+```
+
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 
+the table and hudi will treat them as key less table. Users can still perform Merge Into, updates and deletes based on any random data column. 
+
+### Primary Key
+Here is an example of creating COW table with a primary key 'id'. For mutable datasets, it is recommended to set appropriate primary key. 
+
+```sql
+-- create a managed cow table
+create table if not exists hudi_table0 (
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+);
+```
+
+### PreCombineField
+Here is an example of creating an MOR external table. The **preCombineField** option
+is used to specify the preCombine field for merge. Generally 'event time' or some other similar column will be used for 
+ordering purpose. Hudi will be able to handle out of order data using the precombine field value.
+
+```sql
+-- create an external mor table

Review Comment:
   ditto



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
+
+### Table Type
+Here is an example of creating a COW table.
+
+```sql
+-- create a non-primary key (or key less) table
+create table if not exists hudi_table2(
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow'
+);
+```
+
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 

Review Comment:
   nit: user -> users



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
+
+### Table Type
+Here is an example of creating a COW table.
+
+```sql
+-- create a non-primary key (or key less) table
+create table if not exists hudi_table2(
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow'
+);
+```
+
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 
+the table and hudi will treat them as key less table. Users can still perform Merge Into, updates and deletes based on any random data column. 
+
+### Primary Key
+Here is an example of creating COW table with a primary key 'id'. For mutable datasets, it is recommended to set appropriate primary key. 
+
+```sql
+-- create a managed cow table
+create table if not exists hudi_table0 (
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+);
+```
+
+### PreCombineField
+Here is an example of creating an MOR external table. The **preCombineField** option
+is used to specify the preCombine field for merge. Generally 'event time' or some other similar column will be used for 
+ordering purpose. Hudi will be able to handle out of order data using the precombine field value.
+
+```sql
+-- create an external mor table
+create table if not exists hudi_table1 (
+  id int, 
+  name string, 
+  price double,
+  ts bigint
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id,name',
+  preCombineField = 'ts' 
+);
+```
+
+### Partitioned Table
+Here is an example of creating a COW partitioned key less table.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+ ) 
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table
+Here is an example of creating a COW un-partitioned table.
+
+```sql
+-- create a cow table, with primaryKey 'uuid' and unpartitioned. 
+create table hudi_cow_nonpcf_tbl (
+  uuid int,
+  name string,
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'uuid'
+);
+```
+
+Here is an example of creating a MOR un-partitioned table.
+
+```sql
+-- create a mor non-partitioned table with preCombineField provided
+create table hudi_mor_tbl (
+    id int,
+    name string,
+    price double,
+    ts bigint
+) using hudi
+tblproperties (
+    type = 'mor',
+    primaryKey = 'id',
+    preCombineField = 'ts'
+);
+```
+
+### Create Table for an External Hudi Table
+You can create an External table using the `location` statement. If an external location is not specified it is considered a managed table. 
+You can read more about external vs managed tables [here](https://sparkbyexamples.com/apache-hive/difference-between-hive-internal-tables-and-external-tables/).
+An external table is useful if you need to read/write to/from a pre-existing hudi table.
+
+```sql
+ create table h_p1 using hudi
+ location '/path/to/hudi';
+```
+
+:::tip
+You don't need to specify schema and any properties except the partitioned columns if existed. Hudi can automatically recognize the schema and configurations.
+:::
+
+### Create Table AS SELECT
+
+Hudi supports CTAS(Create table as select) on spark sql. <br/>
+**Note:** For better performance to load data to hudi table, CTAS uses **bulk insert** as the write operation.
+
+**Example CTAS command to create a non-partitioned COW key less table.**
+
+```sql 
+create table h3 using hudi
+tblproperties (type = 'cow')
+as
+select 1 as id, 'a1' as name, 10 as price;
+```
+
+**Example CTAS command to create a partitioned, primary keyed COW table.**
+
+```sql
+create table h2 using hudi
+tblproperties (type = 'cow', primaryKey = 'id')
+partitioned by (dt)
+as
+select 1 as id, 'a1' as name, 10 as price, 1000 as dt;
+```
+
+**Example CTAS command to load data from another table.**
+
+```sql
+# create managed parquet table 
+create table parquet_mngd using parquet location 'file:///tmp/parquet_dataset/*.parquet';
+
+# CTAS by loading data into hudi table
+create table hudi_tbl using hudi location 'file:/tmp/hudi/hudi_tbl/' tblproperties ( 
+  type = 'cow', 
+  primaryKey = 'id', 
+  preCombineField = 'ts' 
+ ) 
+partitioned by (datestr) as select * from parquet_mngd;
+```
+
+### Set hoodie config options
+You can also set the config with table options when creating table which will work for
+the table scope only and override the config set by the SET command.
+```sql
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  ${hoodie.config.key1} = '${hoodie.config.value2}',
+  ${hoodie.config.key2} = '${hoodie.config.value2}',
+  ....
+);
+
+e.g.
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  hoodie.cleaner.fileversions.retained = '20',
+  hoodie.keep.max.commits = '20'
+);
+```
+
+## Spark Alter Table
+### Syntax
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+```
+
+:::note
+`ALTER TABLE ... RENAME TO ...` is not supported when using AWS Glue Data Catalog as hive metastore as Glue itself does 
+not support table renames.
+:::
+
+### Examples
+```sql
+alter table h0 rename to h0_1;
+
+alter table h0_1 add columns(ext0 string);
+
+alter table h0_1 change column id id bigint;
+```
+
+Schema evolution can be achieved via `ALTER TABLE` commands. Below shows some basic examples.
+
+:::note
+For more detailed examples, please prefer to [schema evolution](/docs/schema_evolution)
+:::
+
+**Syntax**
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+
+-- Alter table properties

Review Comment:
   This is already mentioned in a separate section below `### Alter hoodie config options`. We can probably remove it from here.



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
+
+### Table Type
+Here is an example of creating a COW table.
+
+```sql
+-- create a non-primary key (or key less) table
+create table if not exists hudi_table2(
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow'
+);
+```
+
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 
+the table and hudi will treat them as key less table. Users can still perform Merge Into, updates and deletes based on any random data column. 
+
+### Primary Key
+Here is an example of creating COW table with a primary key 'id'. For mutable datasets, it is recommended to set appropriate primary key. 
+
+```sql
+-- create a managed cow table
+create table if not exists hudi_table0 (
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+);
+```
+
+### PreCombineField
+Here is an example of creating an MOR external table. The **preCombineField** option
+is used to specify the preCombine field for merge. Generally 'event time' or some other similar column will be used for 
+ordering purpose. Hudi will be able to handle out of order data using the precombine field value.
+
+```sql
+-- create an external mor table
+create table if not exists hudi_table1 (
+  id int, 
+  name string, 
+  price double,
+  ts bigint
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id,name',
+  preCombineField = 'ts' 
+);
+```
+
+### Partitioned Table
+Here is an example of creating a COW partitioned key less table.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+ ) 
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table
+Here is an example of creating a COW un-partitioned table.
+
+```sql
+-- create a cow table, with primaryKey 'uuid' and unpartitioned. 
+create table hudi_cow_nonpcf_tbl (
+  uuid int,
+  name string,
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'uuid'
+);
+```
+
+Here is an example of creating a MOR un-partitioned table.
+
+```sql
+-- create a mor non-partitioned table with preCombineField provided
+create table hudi_mor_tbl (
+    id int,
+    name string,
+    price double,
+    ts bigint
+) using hudi
+tblproperties (
+    type = 'mor',
+    primaryKey = 'id',
+    preCombineField = 'ts'
+);
+```
+
+### Create Table for an External Hudi Table
+You can create an External table using the `location` statement. If an external location is not specified it is considered a managed table. 
+You can read more about external vs managed tables [here](https://sparkbyexamples.com/apache-hive/difference-between-hive-internal-tables-and-external-tables/).
+An external table is useful if you need to read/write to/from a pre-existing hudi table.
+
+```sql
+ create table h_p1 using hudi
+ location '/path/to/hudi';
+```
+
+:::tip
+You don't need to specify schema and any properties except the partitioned columns if existed. Hudi can automatically recognize the schema and configurations.

Review Comment:
   nit: if existed -> if they exist



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
+
+### Table Type
+Here is an example of creating a COW table.
+
+```sql
+-- create a non-primary key (or key less) table
+create table if not exists hudi_table2(
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow'
+);
+```
+
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 
+the table and hudi will treat them as key less table. Users can still perform Merge Into, updates and deletes based on any random data column. 
+
+### Primary Key
+Here is an example of creating COW table with a primary key 'id'. For mutable datasets, it is recommended to set appropriate primary key. 
+
+```sql
+-- create a managed cow table
+create table if not exists hudi_table0 (
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+);
+```
+
+### PreCombineField
+Here is an example of creating an MOR external table. The **preCombineField** option
+is used to specify the preCombine field for merge. Generally 'event time' or some other similar column will be used for 
+ordering purpose. Hudi will be able to handle out of order data using the precombine field value.
+
+```sql
+-- create an external mor table
+create table if not exists hudi_table1 (
+  id int, 
+  name string, 
+  price double,
+  ts bigint
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id,name',
+  preCombineField = 'ts' 
+);
+```
+
+### Partitioned Table
+Here is an example of creating a COW partitioned key less table.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+ ) 
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table
+Here is an example of creating a COW un-partitioned table.
+
+```sql
+-- create a cow table, with primaryKey 'uuid' and unpartitioned. 
+create table hudi_cow_nonpcf_tbl (
+  uuid int,
+  name string,
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'uuid'
+);
+```
+
+Here is an example of creating a MOR un-partitioned table.
+
+```sql
+-- create a mor non-partitioned table with preCombineField provided
+create table hudi_mor_tbl (
+    id int,
+    name string,
+    price double,
+    ts bigint
+) using hudi
+tblproperties (
+    type = 'mor',
+    primaryKey = 'id',
+    preCombineField = 'ts'
+);
+```
+
+### Create Table for an External Hudi Table
+You can create an External table using the `location` statement. If an external location is not specified it is considered a managed table. 
+You can read more about external vs managed tables [here](https://sparkbyexamples.com/apache-hive/difference-between-hive-internal-tables-and-external-tables/).
+An external table is useful if you need to read/write to/from a pre-existing hudi table.
+
+```sql
+ create table h_p1 using hudi
+ location '/path/to/hudi';
+```
+
+:::tip
+You don't need to specify schema and any properties except the partitioned columns if existed. Hudi can automatically recognize the schema and configurations.
+:::
+
+### Create Table AS SELECT
+
+Hudi supports CTAS(Create table as select) on spark sql. <br/>
+**Note:** For better performance to load data to hudi table, CTAS uses **bulk insert** as the write operation.
+
+**Example CTAS command to create a non-partitioned COW key less table.**
+
+```sql 
+create table h3 using hudi
+tblproperties (type = 'cow')
+as
+select 1 as id, 'a1' as name, 10 as price;
+```
+
+**Example CTAS command to create a partitioned, primary keyed COW table.**
+
+```sql
+create table h2 using hudi
+tblproperties (type = 'cow', primaryKey = 'id')
+partitioned by (dt)
+as
+select 1 as id, 'a1' as name, 10 as price, 1000 as dt;
+```
+
+**Example CTAS command to load data from another table.**
+
+```sql
+# create managed parquet table 
+create table parquet_mngd using parquet location 'file:///tmp/parquet_dataset/*.parquet';
+
+# CTAS by loading data into hudi table
+create table hudi_tbl using hudi location 'file:/tmp/hudi/hudi_tbl/' tblproperties ( 
+  type = 'cow', 
+  primaryKey = 'id', 
+  preCombineField = 'ts' 
+ ) 
+partitioned by (datestr) as select * from parquet_mngd;
+```
+
+### Set hoodie config options
+You can also set the config with table options when creating table which will work for
+the table scope only and override the config set by the SET command.
+```sql
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  ${hoodie.config.key1} = '${hoodie.config.value2}',
+  ${hoodie.config.key2} = '${hoodie.config.value2}',
+  ....
+);
+
+e.g.
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  hoodie.cleaner.fileversions.retained = '20',
+  hoodie.keep.max.commits = '20'
+);
+```
+
+## Spark Alter Table
+### Syntax
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+```
+
+:::note
+`ALTER TABLE ... RENAME TO ...` is not supported when using AWS Glue Data Catalog as hive metastore as Glue itself does 
+not support table renames.
+:::
+
+### Examples
+```sql
+alter table h0 rename to h0_1;
+
+alter table h0_1 add columns(ext0 string);
+
+alter table h0_1 change column id id bigint;
+```
+
+Schema evolution can be achieved via `ALTER TABLE` commands. Below shows some basic examples.
+
+:::note
+For more detailed examples, please prefer to [schema evolution](/docs/schema_evolution)
+:::
+
+**Syntax**
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+
+-- Alter table properties
+ALTER TABLE tableIdentifier SET TBLPROPERTIES (key = 'value')
+```
+
+**Examples**
+```sql
+--rename to:
+ALTER TABLE hudi_cow_nonpcf_tbl RENAME TO hudi_cow_nonpcf_tbl2;
+
+--add column:
+ALTER TABLE hudi_cow_nonpcf_tbl2 add columns(remark string);
+
+--change column:
+ALTER TABLE hudi_cow_nonpcf_tbl2 change column uuid uuid bigint;
+
+--set properties;
+alter table hudi_cow_nonpcf_tbl2 set tblproperties (hoodie.keep.max.commits = '10');
+```
+
+### Alter hoodie config options
+You can also alter the write config for a table by the **ALTER SERDEPROPERTIES**

Review Comment:
   nit: ALTER SERDEPROPERTIES -> ALTER TABLE set serdeproperties



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
+
+### Table Type
+Here is an example of creating a COW table.
+
+```sql
+-- create a non-primary key (or key less) table
+create table if not exists hudi_table2(
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow'
+);
+```
+
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 
+the table and hudi will treat them as key less table. Users can still perform Merge Into, updates and deletes based on any random data column. 
+
+### Primary Key
+Here is an example of creating COW table with a primary key 'id'. For mutable datasets, it is recommended to set appropriate primary key. 
+
+```sql
+-- create a managed cow table
+create table if not exists hudi_table0 (
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+);
+```
+
+### PreCombineField
+Here is an example of creating an MOR external table. The **preCombineField** option
+is used to specify the preCombine field for merge. Generally 'event time' or some other similar column will be used for 
+ordering purpose. Hudi will be able to handle out of order data using the precombine field value.
+
+```sql
+-- create an external mor table
+create table if not exists hudi_table1 (
+  id int, 
+  name string, 
+  price double,
+  ts bigint
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id,name',
+  preCombineField = 'ts' 
+);
+```
+
+### Partitioned Table
+Here is an example of creating a COW partitioned key less table.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+ ) 
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table
+Here is an example of creating a COW un-partitioned table.
+
+```sql
+-- create a cow table, with primaryKey 'uuid' and unpartitioned. 
+create table hudi_cow_nonpcf_tbl (
+  uuid int,
+  name string,
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'uuid'
+);
+```
+
+Here is an example of creating a MOR un-partitioned table.
+
+```sql
+-- create a mor non-partitioned table with preCombineField provided
+create table hudi_mor_tbl (
+    id int,
+    name string,
+    price double,
+    ts bigint
+) using hudi
+tblproperties (
+    type = 'mor',
+    primaryKey = 'id',
+    preCombineField = 'ts'
+);
+```
+
+### Create Table for an External Hudi Table
+You can create an External table using the `location` statement. If an external location is not specified it is considered a managed table. 
+You can read more about external vs managed tables [here](https://sparkbyexamples.com/apache-hive/difference-between-hive-internal-tables-and-external-tables/).
+An external table is useful if you need to read/write to/from a pre-existing hudi table.
+
+```sql
+ create table h_p1 using hudi
+ location '/path/to/hudi';
+```
+
+:::tip
+You don't need to specify schema and any properties except the partitioned columns if existed. Hudi can automatically recognize the schema and configurations.
+:::
+
+### Create Table AS SELECT
+
+Hudi supports CTAS(Create table as select) on spark sql. <br/>
+**Note:** For better performance to load data to hudi table, CTAS uses **bulk insert** as the write operation.
+
+**Example CTAS command to create a non-partitioned COW key less table.**
+
+```sql 
+create table h3 using hudi
+tblproperties (type = 'cow')
+as
+select 1 as id, 'a1' as name, 10 as price;
+```
+
+**Example CTAS command to create a partitioned, primary keyed COW table.**
+
+```sql
+create table h2 using hudi
+tblproperties (type = 'cow', primaryKey = 'id')
+partitioned by (dt)
+as
+select 1 as id, 'a1' as name, 10 as price, 1000 as dt;
+```
+
+**Example CTAS command to load data from another table.**
+
+```sql
+# create managed parquet table 
+create table parquet_mngd using parquet location 'file:///tmp/parquet_dataset/*.parquet';
+
+# CTAS by loading data into hudi table
+create table hudi_tbl using hudi location 'file:/tmp/hudi/hudi_tbl/' tblproperties ( 
+  type = 'cow', 
+  primaryKey = 'id', 
+  preCombineField = 'ts' 
+ ) 
+partitioned by (datestr) as select * from parquet_mngd;
+```
+
+### Set hoodie config options
+You can also set the config with table options when creating table which will work for
+the table scope only and override the config set by the SET command.
+```sql
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  ${hoodie.config.key1} = '${hoodie.config.value2}',
+  ${hoodie.config.key2} = '${hoodie.config.value2}',
+  ....
+);
+
+e.g.
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  hoodie.cleaner.fileversions.retained = '20',
+  hoodie.keep.max.commits = '20'
+);
+```
+
+## Spark Alter Table
+### Syntax
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)

Review Comment:
   nit: [, colAndType]



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
+
+### Table Type
+Here is an example of creating a COW table.
+
+```sql
+-- create a non-primary key (or key less) table
+create table if not exists hudi_table2(
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow'
+);
+```
+
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 
+the table and hudi will treat them as key less table. Users can still perform Merge Into, updates and deletes based on any random data column. 
+
+### Primary Key
+Here is an example of creating COW table with a primary key 'id'. For mutable datasets, it is recommended to set appropriate primary key. 
+
+```sql
+-- create a managed cow table
+create table if not exists hudi_table0 (
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+);
+```
+
+### PreCombineField
+Here is an example of creating an MOR external table. The **preCombineField** option
+is used to specify the preCombine field for merge. Generally 'event time' or some other similar column will be used for 
+ordering purpose. Hudi will be able to handle out of order data using the precombine field value.
+
+```sql
+-- create an external mor table
+create table if not exists hudi_table1 (
+  id int, 
+  name string, 
+  price double,
+  ts bigint
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id,name',
+  preCombineField = 'ts' 
+);
+```
+
+### Partitioned Table
+Here is an example of creating a COW partitioned key less table.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+ ) 
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table
+Here is an example of creating a COW un-partitioned table.
+
+```sql
+-- create a cow table, with primaryKey 'uuid' and unpartitioned. 
+create table hudi_cow_nonpcf_tbl (
+  uuid int,
+  name string,
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'uuid'
+);
+```
+
+Here is an example of creating a MOR un-partitioned table.
+
+```sql
+-- create a mor non-partitioned table with preCombineField provided
+create table hudi_mor_tbl (
+    id int,
+    name string,
+    price double,
+    ts bigint
+) using hudi
+tblproperties (
+    type = 'mor',
+    primaryKey = 'id',
+    preCombineField = 'ts'
+);
+```
+
+### Create Table for an External Hudi Table
+You can create an External table using the `location` statement. If an external location is not specified it is considered a managed table. 
+You can read more about external vs managed tables [here](https://sparkbyexamples.com/apache-hive/difference-between-hive-internal-tables-and-external-tables/).
+An external table is useful if you need to read/write to/from a pre-existing hudi table.
+
+```sql
+ create table h_p1 using hudi
+ location '/path/to/hudi';
+```
+
+:::tip
+You don't need to specify schema and any properties except the partitioned columns if existed. Hudi can automatically recognize the schema and configurations.
+:::
+
+### Create Table AS SELECT
+
+Hudi supports CTAS(Create table as select) on spark sql. <br/>
+**Note:** For better performance to load data to hudi table, CTAS uses **bulk insert** as the write operation.
+
+**Example CTAS command to create a non-partitioned COW key less table.**
+
+```sql 
+create table h3 using hudi
+tblproperties (type = 'cow')
+as
+select 1 as id, 'a1' as name, 10 as price;
+```
+
+**Example CTAS command to create a partitioned, primary keyed COW table.**
+
+```sql
+create table h2 using hudi
+tblproperties (type = 'cow', primaryKey = 'id')
+partitioned by (dt)
+as
+select 1 as id, 'a1' as name, 10 as price, 1000 as dt;
+```
+
+**Example CTAS command to load data from another table.**
+
+```sql
+# create managed parquet table 
+create table parquet_mngd using parquet location 'file:///tmp/parquet_dataset/*.parquet';
+
+# CTAS by loading data into hudi table
+create table hudi_tbl using hudi location 'file:/tmp/hudi/hudi_tbl/' tblproperties ( 
+  type = 'cow', 
+  primaryKey = 'id', 
+  preCombineField = 'ts' 
+ ) 
+partitioned by (datestr) as select * from parquet_mngd;
+```
+
+### Set hoodie config options
+You can also set the config with table options when creating table which will work for
+the table scope only and override the config set by the SET command.
+```sql
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  ${hoodie.config.key1} = '${hoodie.config.value2}',
+  ${hoodie.config.key2} = '${hoodie.config.value2}',
+  ....
+);
+
+e.g.
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  hoodie.cleaner.fileversions.retained = '20',
+  hoodie.keep.max.commits = '20'
+);
+```
+
+## Spark Alter Table
+### Syntax
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+```
+
+:::note
+`ALTER TABLE ... RENAME TO ...` is not supported when using AWS Glue Data Catalog as hive metastore as Glue itself does 
+not support table renames.
+:::
+
+### Examples
+```sql
+alter table h0 rename to h0_1;
+
+alter table h0_1 add columns(ext0 string);
+
+alter table h0_1 change column id id bigint;
+```
+
+Schema evolution can be achieved via `ALTER TABLE` commands. Below shows some basic examples.
+
+:::note
+For more detailed examples, please prefer to [schema evolution](/docs/schema_evolution)
+:::
+
+**Syntax**
+```sql

Review Comment:
   All this is already covered above, can be removed. We can just keep the note markdown mentioning about schema evolution and remove the rest of duplicate stuff.



##########
website/docs/sql_dml.md:
##########
@@ -7,42 +7,46 @@ last_modified_at:
 import Tabs from '@theme/Tabs';
 import TabItem from '@theme/TabItem';
 
-The following are SparkSQL table management actions available:
+The following are SparkSQL DDL actions available:
 
 ## Spark Create Table
 :::note
-Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
 :::
 
 ### Options
 
 Users can set table options while creating a hudi table.
 
-| Parameter Name | Description | (Optional/Required) : Default Value |
-|------------|--------|--------|
-| primaryKey | The primary key names of the table, multiple fields separated by commas. | (Optional) : `id`|
-| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.| (Optional) : `cow` |
-| preCombineField | The Pre-Combine field of the table. | (Optional) : `ts`|
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|

Review Comment:
   nit: two -> multiple



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
+
+### Table Type
+Here is an example of creating a COW table.
+
+```sql
+-- create a non-primary key (or key less) table
+create table if not exists hudi_table2(
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow'
+);
+```
+
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 
+the table and hudi will treat them as key less table. Users can still perform Merge Into, updates and deletes based on any random data column. 
+
+### Primary Key
+Here is an example of creating COW table with a primary key 'id'. For mutable datasets, it is recommended to set appropriate primary key. 
+
+```sql
+-- create a managed cow table
+create table if not exists hudi_table0 (
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+);
+```
+
+### PreCombineField
+Here is an example of creating an MOR external table. The **preCombineField** option

Review Comment:
   The example does not create external table. Please correct me if I am wrong.



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
+
+### Table Type
+Here is an example of creating a COW table.
+
+```sql
+-- create a non-primary key (or key less) table
+create table if not exists hudi_table2(
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow'
+);
+```
+
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 
+the table and hudi will treat them as key less table. Users can still perform Merge Into, updates and deletes based on any random data column. 
+
+### Primary Key
+Here is an example of creating COW table with a primary key 'id'. For mutable datasets, it is recommended to set appropriate primary key. 
+
+```sql
+-- create a managed cow table
+create table if not exists hudi_table0 (
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+);
+```
+
+### PreCombineField
+Here is an example of creating an MOR external table. The **preCombineField** option
+is used to specify the preCombine field for merge. Generally 'event time' or some other similar column will be used for 
+ordering purpose. Hudi will be able to handle out of order data using the precombine field value.
+
+```sql
+-- create an external mor table
+create table if not exists hudi_table1 (
+  id int, 
+  name string, 
+  price double,
+  ts bigint
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id,name',
+  preCombineField = 'ts' 
+);
+```
+
+### Partitioned Table
+Here is an example of creating a COW partitioned key less table.

Review Comment:
   primaryKey is specified in the sql statement.



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
+
+### Table Type
+Here is an example of creating a COW table.
+
+```sql
+-- create a non-primary key (or key less) table
+create table if not exists hudi_table2(
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow'
+);
+```
+
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 
+the table and hudi will treat them as key less table. Users can still perform Merge Into, updates and deletes based on any random data column. 
+
+### Primary Key
+Here is an example of creating COW table with a primary key 'id'. For mutable datasets, it is recommended to set appropriate primary key. 
+
+```sql
+-- create a managed cow table
+create table if not exists hudi_table0 (
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+);
+```
+
+### PreCombineField
+Here is an example of creating an MOR external table. The **preCombineField** option
+is used to specify the preCombine field for merge. Generally 'event time' or some other similar column will be used for 
+ordering purpose. Hudi will be able to handle out of order data using the precombine field value.
+
+```sql
+-- create an external mor table
+create table if not exists hudi_table1 (
+  id int, 
+  name string, 
+  price double,
+  ts bigint
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id,name',
+  preCombineField = 'ts' 
+);
+```
+
+### Partitioned Table
+Here is an example of creating a COW partitioned key less table.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+ ) 
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table
+Here is an example of creating a COW un-partitioned table.

Review Comment:
   ditto



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
+
+### Table Type
+Here is an example of creating a COW table.
+
+```sql
+-- create a non-primary key (or key less) table
+create table if not exists hudi_table2(
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow'
+);
+```
+
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 
+the table and hudi will treat them as key less table. Users can still perform Merge Into, updates and deletes based on any random data column. 
+
+### Primary Key
+Here is an example of creating COW table with a primary key 'id'. For mutable datasets, it is recommended to set appropriate primary key. 
+
+```sql
+-- create a managed cow table
+create table if not exists hudi_table0 (
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+);
+```
+
+### PreCombineField
+Here is an example of creating an MOR external table. The **preCombineField** option
+is used to specify the preCombine field for merge. Generally 'event time' or some other similar column will be used for 
+ordering purpose. Hudi will be able to handle out of order data using the precombine field value.
+
+```sql
+-- create an external mor table
+create table if not exists hudi_table1 (
+  id int, 
+  name string, 
+  price double,
+  ts bigint
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id,name',
+  preCombineField = 'ts' 
+);
+```
+
+### Partitioned Table
+Here is an example of creating a COW partitioned key less table.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+ ) 
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table
+Here is an example of creating a COW un-partitioned table.
+
+```sql
+-- create a cow table, with primaryKey 'uuid' and unpartitioned. 
+create table hudi_cow_nonpcf_tbl (
+  uuid int,
+  name string,
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'uuid'
+);
+```
+
+Here is an example of creating a MOR un-partitioned table.

Review Comment:
   ditto



##########
website/docs/sql_dml.md:
##########
@@ -69,56 +74,110 @@ create table if not exists hudi_table1 (
   price double,
   ts bigint
 ) using hudi
-options (
+tblproperties (
   type = 'mor',
   primaryKey = 'id,name',
   preCombineField = 'ts' 
 );
 ```
 
 ### Partitioned Table
-Here is an example of creating a COW partitioned table.
+Here is an example of creating a COW partitioned key less table.
 ```sql
 create table if not exists hudi_table_p0 (
 id bigint,
 name string,
 dt string,
 hh string  
 ) using hudi
-options (
+tblproperties (
   type = 'cow',
   primaryKey = 'id'
  ) 
-partitioned by (dt, hh);
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table

Review Comment:
   nit: Non-Partitioned



##########
website/docs/sql_dml.md:
##########
@@ -7,42 +7,46 @@ last_modified_at:
 import Tabs from '@theme/Tabs';
 import TabItem from '@theme/TabItem';
 
-The following are SparkSQL table management actions available:
+The following are SparkSQL DDL actions available:
 
 ## Spark Create Table
 :::note
-Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
 :::
 
 ### Options
 
 Users can set table options while creating a hudi table.
 
-| Parameter Name | Description | (Optional/Required) : Default Value |
-|------------|--------|--------|
-| primaryKey | The primary key names of the table, multiple fields separated by commas. | (Optional) : `id`|
-| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.| (Optional) : `cow` |
-| preCombineField | The Pre-Combine field of the table. | (Optional) : `ts`|
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|

Review Comment:
   Ideally the default value should be None for primaryKey and preCombineField, since it is unique for every use case. What do you think? @nsivabalan 



##########
website/docs/quick-start-guide.md:
##########
@@ -216,9 +237,33 @@ can generate sample inserts and updates based on the the sample trip schema [her
 
 ## Create Table
 
+Before we go further, few terminologies to familiarize: 

Review Comment:
   nit: "few terminologies to familiarize:" -> "let us go over few terminologies:"



##########
website/docs/quick-start-guide.md:
##########
@@ -245,72 +290,48 @@ values={[
 
 Spark SQL needs an explicit create table command.
 
-**Table Concepts**
-
-- Table types
-
-  Both Hudi's table types, Copy-On-Write (COW) and Merge-On-Read (MOR), can be created using Spark SQL.
-  While creating the table, table type can be specified using **type** option: **type = 'cow'** or **type = 'mor'**.
-
-- Partitioned & Non-Partitioned tables
-
-  Users can create a partitioned table or a non-partitioned table in Spark SQL. To create a partitioned table, one needs
-  to use **partitioned by** statement to specify the partition columns to create a partitioned table. When there is
-  no **partitioned by** statement with create table command, table is considered to be a non-partitioned table.
-
-- Managed & External tables
+**Create Table Properties**
 
-  In general, Spark SQL supports two kinds of tables, namely managed and external. If one specifies a location using **
-  location** statement or use `create external table` to create table explicitly, it is an external table, else its
-  considered a managed table. You can read more about external vs managed
-  tables [here](https://sparkbyexamples.com/apache-hive/difference-between-hive-internal-tables-and-external-tables/).
+Users can set table properties while creating a hudi table. Critical options are listed here.
 
-*Read more in the [table management](/docs/table_management) guide.*
+| Parameter Name | Default | Introduction                                                                                                                                                                                                                                                                                                                                  |
+|------------------|--------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| primaryKey | uuid | The primary key field names of the table, multiple fields separated by commas. Same as `hoodie.datasource.write.recordkey.field`. This can be ignored for a key less table (from 0.14.0).                                                                                                                                                     |
+| preCombineField |  | The pre-combine field of the table. Same as `hoodie.datasource.write.precombine.field`. is used for resolving the final version of the record among multiple versions. Generally `event time` or some other similar column will be used for ordering purpose. Hudi will be able to handle out of order data using the preCombine field value. |

Review Comment:
   nit: "This" is used for resolving



##########
website/docs/sql_dml.md:
##########
@@ -69,56 +74,110 @@ create table if not exists hudi_table1 (
   price double,
   ts bigint
 ) using hudi
-options (
+tblproperties (
   type = 'mor',
   primaryKey = 'id,name',
   preCombineField = 'ts' 
 );
 ```
 
 ### Partitioned Table
-Here is an example of creating a COW partitioned table.
+Here is an example of creating a COW partitioned key less table.
 ```sql
 create table if not exists hudi_table_p0 (
 id bigint,
 name string,
 dt string,
 hh string  
 ) using hudi
-options (
+tblproperties (
   type = 'cow',
   primaryKey = 'id'
  ) 
-partitioned by (dt, hh);
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table
+Here is an example of creating a COW un-partitioned table.
+
+```sql
+-- create a cow table, with primaryKey 'uuid' and unpartitioned. 
+create table hudi_cow_nonpcf_tbl (
+  uuid int,
+  name string,
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'uuid'
+);
+```
+
+Here is an example of creating a MOR un-partitioned table.
+
+```sql
+-- create a mor non-partitioned table with preCombineField provided
+create table hudi_mor_tbl (
+    id int,
+    name string,
+    price double,
+    ts bigint
+) using hudi
+tblproperties (
+    type = 'mor',
+    primaryKey = 'id',
+    preCombineField = 'ts'
+);
 ```
 
 ### Create Table for an External Hudi Table
-You can create an External table using the `location` statement. If an external location is not specified it is considered a managed table. You can read more about external vs managed tables [here](https://sparkbyexamples.com/apache-hive/difference-between-hive-internal-tables-and-external-tables/).
+You can create an External table using the `location` statement. If an external location is not specified it is considered a managed table. 
+You can read more about external vs managed tables [here](https://sparkbyexamples.com/apache-hive/difference-between-hive-internal-tables-and-external-tables/).
 An external table is useful if you need to read/write to/from a pre-existing hudi table.
 
 ```sql
  create table h_p1 using hudi
  location '/path/to/hudi';
 ```
 
+:::tip
+You don't need to specify schema and any properties except the partitioned columns if existed. Hudi can automatically recognize the schema and configurations.

Review Comment:
   nit: if existed -> if they exist



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
+
+### Table Type
+Here is an example of creating a COW table.
+
+```sql
+-- create a non-primary key (or key less) table
+create table if not exists hudi_table2(
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow'
+);
+```
+
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 
+the table and hudi will treat them as key less table. Users can still perform Merge Into, updates and deletes based on any random data column. 
+
+### Primary Key
+Here is an example of creating COW table with a primary key 'id'. For mutable datasets, it is recommended to set appropriate primary key. 
+
+```sql
+-- create a managed cow table
+create table if not exists hudi_table0 (
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+);
+```
+
+### PreCombineField
+Here is an example of creating an MOR external table. The **preCombineField** option
+is used to specify the preCombine field for merge. Generally 'event time' or some other similar column will be used for 
+ordering purpose. Hudi will be able to handle out of order data using the precombine field value.
+
+```sql
+-- create an external mor table
+create table if not exists hudi_table1 (
+  id int, 
+  name string, 
+  price double,
+  ts bigint
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id,name',
+  preCombineField = 'ts' 
+);
+```
+
+### Partitioned Table
+Here is an example of creating a COW partitioned key less table.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+ ) 
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table
+Here is an example of creating a COW un-partitioned table.
+
+```sql
+-- create a cow table, with primaryKey 'uuid' and unpartitioned. 
+create table hudi_cow_nonpcf_tbl (
+  uuid int,
+  name string,
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'uuid'
+);
+```
+
+Here is an example of creating a MOR un-partitioned table.
+
+```sql
+-- create a mor non-partitioned table with preCombineField provided
+create table hudi_mor_tbl (
+    id int,
+    name string,
+    price double,
+    ts bigint
+) using hudi
+tblproperties (
+    type = 'mor',
+    primaryKey = 'id',
+    preCombineField = 'ts'
+);
+```
+
+### Create Table for an External Hudi Table
+You can create an External table using the `location` statement. If an external location is not specified it is considered a managed table. 
+You can read more about external vs managed tables [here](https://sparkbyexamples.com/apache-hive/difference-between-hive-internal-tables-and-external-tables/).
+An external table is useful if you need to read/write to/from a pre-existing hudi table.
+
+```sql
+ create table h_p1 using hudi
+ location '/path/to/hudi';
+```
+
+:::tip
+You don't need to specify schema and any properties except the partitioned columns if existed. Hudi can automatically recognize the schema and configurations.
+:::
+
+### Create Table AS SELECT
+
+Hudi supports CTAS(Create table as select) on spark sql. <br/>
+**Note:** For better performance to load data to hudi table, CTAS uses **bulk insert** as the write operation.
+
+**Example CTAS command to create a non-partitioned COW key less table.**
+
+```sql 
+create table h3 using hudi
+tblproperties (type = 'cow')
+as
+select 1 as id, 'a1' as name, 10 as price;
+```
+
+**Example CTAS command to create a partitioned, primary keyed COW table.**
+
+```sql
+create table h2 using hudi
+tblproperties (type = 'cow', primaryKey = 'id')
+partitioned by (dt)
+as
+select 1 as id, 'a1' as name, 10 as price, 1000 as dt;
+```
+
+**Example CTAS command to load data from another table.**
+
+```sql
+# create managed parquet table 
+create table parquet_mngd using parquet location 'file:///tmp/parquet_dataset/*.parquet';
+
+# CTAS by loading data into hudi table
+create table hudi_tbl using hudi location 'file:/tmp/hudi/hudi_tbl/' tblproperties ( 
+  type = 'cow', 
+  primaryKey = 'id', 
+  preCombineField = 'ts' 
+ ) 
+partitioned by (datestr) as select * from parquet_mngd;
+```
+
+### Set hoodie config options
+You can also set the config with table options when creating table which will work for
+the table scope only and override the config set by the SET command.
+```sql
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  ${hoodie.config.key1} = '${hoodie.config.value2}',
+  ${hoodie.config.key2} = '${hoodie.config.value2}',
+  ....
+);
+
+e.g.
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  hoodie.cleaner.fileversions.retained = '20',
+  hoodie.keep.max.commits = '20'
+);
+```
+
+## Spark Alter Table
+### Syntax
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+```
+
+:::note
+`ALTER TABLE ... RENAME TO ...` is not supported when using AWS Glue Data Catalog as hive metastore as Glue itself does 

Review Comment:
   nit: as hive metastore -> as the metastore



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
+
+### Table Type
+Here is an example of creating a COW table.
+
+```sql
+-- create a non-primary key (or key less) table
+create table if not exists hudi_table2(
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow'
+);
+```
+
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 
+the table and hudi will treat them as key less table. Users can still perform Merge Into, updates and deletes based on any random data column. 
+
+### Primary Key
+Here is an example of creating COW table with a primary key 'id'. For mutable datasets, it is recommended to set appropriate primary key. 
+
+```sql
+-- create a managed cow table
+create table if not exists hudi_table0 (
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+);
+```
+
+### PreCombineField
+Here is an example of creating an MOR external table. The **preCombineField** option
+is used to specify the preCombine field for merge. Generally 'event time' or some other similar column will be used for 
+ordering purpose. Hudi will be able to handle out of order data using the precombine field value.
+
+```sql
+-- create an external mor table
+create table if not exists hudi_table1 (
+  id int, 
+  name string, 
+  price double,
+  ts bigint
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id,name',
+  preCombineField = 'ts' 
+);
+```
+
+### Partitioned Table
+Here is an example of creating a COW partitioned key less table.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+ ) 
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table
+Here is an example of creating a COW un-partitioned table.
+
+```sql
+-- create a cow table, with primaryKey 'uuid' and unpartitioned. 
+create table hudi_cow_nonpcf_tbl (
+  uuid int,
+  name string,
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'uuid'
+);
+```
+
+Here is an example of creating a MOR un-partitioned table.
+
+```sql
+-- create a mor non-partitioned table with preCombineField provided
+create table hudi_mor_tbl (
+    id int,
+    name string,
+    price double,
+    ts bigint
+) using hudi
+tblproperties (
+    type = 'mor',
+    primaryKey = 'id',
+    preCombineField = 'ts'
+);
+```
+
+### Create Table for an External Hudi Table
+You can create an External table using the `location` statement. If an external location is not specified it is considered a managed table. 
+You can read more about external vs managed tables [here](https://sparkbyexamples.com/apache-hive/difference-between-hive-internal-tables-and-external-tables/).
+An external table is useful if you need to read/write to/from a pre-existing hudi table.
+
+```sql
+ create table h_p1 using hudi
+ location '/path/to/hudi';
+```
+
+:::tip
+You don't need to specify schema and any properties except the partitioned columns if existed. Hudi can automatically recognize the schema and configurations.
+:::
+
+### Create Table AS SELECT
+
+Hudi supports CTAS(Create table as select) on spark sql. <br/>
+**Note:** For better performance to load data to hudi table, CTAS uses **bulk insert** as the write operation.
+
+**Example CTAS command to create a non-partitioned COW key less table.**
+
+```sql 
+create table h3 using hudi
+tblproperties (type = 'cow')
+as
+select 1 as id, 'a1' as name, 10 as price;
+```
+
+**Example CTAS command to create a partitioned, primary keyed COW table.**
+
+```sql
+create table h2 using hudi
+tblproperties (type = 'cow', primaryKey = 'id')
+partitioned by (dt)
+as
+select 1 as id, 'a1' as name, 10 as price, 1000 as dt;
+```
+
+**Example CTAS command to load data from another table.**
+
+```sql
+# create managed parquet table 
+create table parquet_mngd using parquet location 'file:///tmp/parquet_dataset/*.parquet';
+
+# CTAS by loading data into hudi table
+create table hudi_tbl using hudi location 'file:/tmp/hudi/hudi_tbl/' tblproperties ( 
+  type = 'cow', 
+  primaryKey = 'id', 
+  preCombineField = 'ts' 
+ ) 
+partitioned by (datestr) as select * from parquet_mngd;
+```
+
+### Set hoodie config options
+You can also set the config with table options when creating table which will work for
+the table scope only and override the config set by the SET command.
+```sql
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  ${hoodie.config.key1} = '${hoodie.config.value2}',
+  ${hoodie.config.key2} = '${hoodie.config.value2}',
+  ....
+);
+
+e.g.
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  hoodie.cleaner.fileversions.retained = '20',
+  hoodie.keep.max.commits = '20'
+);
+```
+
+## Spark Alter Table
+### Syntax
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+```
+
+:::note
+`ALTER TABLE ... RENAME TO ...` is not supported when using AWS Glue Data Catalog as hive metastore as Glue itself does 
+not support table renames.
+:::
+
+### Examples
+```sql
+alter table h0 rename to h0_1;
+
+alter table h0_1 add columns(ext0 string);
+
+alter table h0_1 change column id id bigint;
+```
+
+Schema evolution can be achieved via `ALTER TABLE` commands. Below shows some basic examples.
+
+:::note
+For more detailed examples, please prefer to [schema evolution](/docs/schema_evolution)
+:::
+
+**Syntax**
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+
+-- Alter table properties
+ALTER TABLE tableIdentifier SET TBLPROPERTIES (key = 'value')
+```
+
+**Examples**
+```sql
+--rename to:
+ALTER TABLE hudi_cow_nonpcf_tbl RENAME TO hudi_cow_nonpcf_tbl2;
+
+--add column:
+ALTER TABLE hudi_cow_nonpcf_tbl2 add columns(remark string);
+
+--change column:
+ALTER TABLE hudi_cow_nonpcf_tbl2 change column uuid uuid bigint;
+
+--set properties;
+alter table hudi_cow_nonpcf_tbl2 set tblproperties (hoodie.keep.max.commits = '10');
+```
+
+### Alter hoodie config options
+You can also alter the write config for a table by the **ALTER SERDEPROPERTIES**
+
+Example:
+```sql
+ alter table h3 set serdeproperties (hoodie.keep.max.commits = '10') 

Review Comment:
   Just trying to understand, these configs can be altered using `set tblproperties` and `set serdeproperties` both?



##########
website/docs/quick-start-guide.md:
##########
@@ -778,18 +727,20 @@ val updates = convertToStringList(dataGen.generateUpdates(10))
 val df = spark.read.json(spark.sparkContext.parallelize(updates, 2))
 df.write.format("hudi").
   options(getQuickstartWriteConfigs).
-  option(PRECOMBINE_FIELD_OPT_KEY, "ts").
-  option(RECORDKEY_FIELD_OPT_KEY, "uuid").
-  option(PARTITIONPATH_FIELD_OPT_KEY, "partitionpath").
+  option(PRECOMBINE_FIELD_NAME.key(), "ts").
+  option(RECORDKEY_FIELD_NAME.key(), "uuid").
+  option(PARTITIONPATH_FIELD_NAME.key(), "partitionpath").
   option(TABLE_NAME, tableName).
   mode(Append).
   save(basePath)
 ```
 :::note
-Notice that the save mode is now `Append`. In general, always use append mode unless you are trying to create the table for the first time.
-[Querying](#query-data) the data again will now show updated trips. Each write operation generates a new [commit](/docs/concepts)
+- Notice that the save mode is now `Append`. In general, always use append mode unless you are trying to create the table for the first time.
+[Querying](#query-data) the data again will now show updated trips. Each write operation generates a new [commit](/docs/next/concepts)
 denoted by the timestamp. Look for changes in `_hoodie_commit_time`, `rider`, `driver` fields for the same `_hoodie_record_key`s in previous commit.
+- Updates to a Keyless table via spark datasource is not supported. Updates for keyless tables are supported with Spark-SQL MergeInto and UPDATE commands.

Review Comment:
   Just curious, I guess updates to keyless table via deltaStreamer are also not supported in that case?



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
+
+### Table Type
+Here is an example of creating a COW table.
+
+```sql
+-- create a non-primary key (or key less) table
+create table if not exists hudi_table2(
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow'
+);
+```
+
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 
+the table and hudi will treat them as key less table. Users can still perform Merge Into, updates and deletes based on any random data column. 
+
+### Primary Key
+Here is an example of creating COW table with a primary key 'id'. For mutable datasets, it is recommended to set appropriate primary key. 
+
+```sql
+-- create a managed cow table
+create table if not exists hudi_table0 (
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+);
+```
+
+### PreCombineField
+Here is an example of creating an MOR external table. The **preCombineField** option
+is used to specify the preCombine field for merge. Generally 'event time' or some other similar column will be used for 
+ordering purpose. Hudi will be able to handle out of order data using the precombine field value.
+
+```sql
+-- create an external mor table
+create table if not exists hudi_table1 (
+  id int, 
+  name string, 
+  price double,
+  ts bigint
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id,name',
+  preCombineField = 'ts' 
+);
+```
+
+### Partitioned Table
+Here is an example of creating a COW partitioned key less table.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+ ) 
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table
+Here is an example of creating a COW un-partitioned table.
+
+```sql
+-- create a cow table, with primaryKey 'uuid' and unpartitioned. 
+create table hudi_cow_nonpcf_tbl (
+  uuid int,
+  name string,
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'uuid'
+);
+```
+
+Here is an example of creating a MOR un-partitioned table.
+
+```sql
+-- create a mor non-partitioned table with preCombineField provided
+create table hudi_mor_tbl (
+    id int,
+    name string,
+    price double,
+    ts bigint
+) using hudi
+tblproperties (
+    type = 'mor',
+    primaryKey = 'id',
+    preCombineField = 'ts'
+);
+```
+
+### Create Table for an External Hudi Table
+You can create an External table using the `location` statement. If an external location is not specified it is considered a managed table. 
+You can read more about external vs managed tables [here](https://sparkbyexamples.com/apache-hive/difference-between-hive-internal-tables-and-external-tables/).
+An external table is useful if you need to read/write to/from a pre-existing hudi table.
+
+```sql
+ create table h_p1 using hudi
+ location '/path/to/hudi';
+```
+
+:::tip
+You don't need to specify schema and any properties except the partitioned columns if existed. Hudi can automatically recognize the schema and configurations.
+:::
+
+### Create Table AS SELECT
+
+Hudi supports CTAS(Create table as select) on spark sql. <br/>
+**Note:** For better performance to load data to hudi table, CTAS uses **bulk insert** as the write operation.
+
+**Example CTAS command to create a non-partitioned COW key less table.**
+
+```sql 
+create table h3 using hudi
+tblproperties (type = 'cow')
+as
+select 1 as id, 'a1' as name, 10 as price;
+```
+
+**Example CTAS command to create a partitioned, primary keyed COW table.**
+
+```sql
+create table h2 using hudi
+tblproperties (type = 'cow', primaryKey = 'id')
+partitioned by (dt)
+as
+select 1 as id, 'a1' as name, 10 as price, 1000 as dt;
+```
+
+**Example CTAS command to load data from another table.**
+
+```sql
+# create managed parquet table 
+create table parquet_mngd using parquet location 'file:///tmp/parquet_dataset/*.parquet';
+
+# CTAS by loading data into hudi table
+create table hudi_tbl using hudi location 'file:/tmp/hudi/hudi_tbl/' tblproperties ( 
+  type = 'cow', 
+  primaryKey = 'id', 
+  preCombineField = 'ts' 
+ ) 
+partitioned by (datestr) as select * from parquet_mngd;
+```
+
+### Set hoodie config options
+You can also set the config with table options when creating table which will work for
+the table scope only and override the config set by the SET command.
+```sql
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  ${hoodie.config.key1} = '${hoodie.config.value2}',
+  ${hoodie.config.key2} = '${hoodie.config.value2}',
+  ....
+);
+
+e.g.
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  hoodie.cleaner.fileversions.retained = '20',
+  hoodie.keep.max.commits = '20'
+);
+```
+
+## Spark Alter Table
+### Syntax
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+```
+
+:::note
+`ALTER TABLE ... RENAME TO ...` is not supported when using AWS Glue Data Catalog as hive metastore as Glue itself does 
+not support table renames.
+:::
+
+### Examples
+```sql
+alter table h0 rename to h0_1;
+
+alter table h0_1 add columns(ext0 string);
+
+alter table h0_1 change column id id bigint;
+```
+
+Schema evolution can be achieved via `ALTER TABLE` commands. Below shows some basic examples.
+
+:::note
+For more detailed examples, please prefer to [schema evolution](/docs/schema_evolution)
+:::
+
+**Syntax**
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+
+-- Alter table properties
+ALTER TABLE tableIdentifier SET TBLPROPERTIES (key = 'value')
+```
+
+**Examples**
+```sql
+--rename to:
+ALTER TABLE hudi_cow_nonpcf_tbl RENAME TO hudi_cow_nonpcf_tbl2;
+
+--add column:
+ALTER TABLE hudi_cow_nonpcf_tbl2 add columns(remark string);
+
+--change column:
+ALTER TABLE hudi_cow_nonpcf_tbl2 change column uuid uuid bigint;
+
+--set properties;

Review Comment:
   This is covered in a separate section below. We can remove it from here to avoid duplication.



##########
website/docs/quick-start-guide.md:
##########
@@ -530,20 +566,24 @@ df.write.format("hudi"). \
     mode("overwrite"). \
     save(basePath)
 ```
+
 :::info
 `mode(Overwrite)` overwrites and recreates the table if it already exists.
 You can check the data generated under `/tmp/hudi_trips_cow/<region>/<country>/<city>/`. We provided a record key
 (`uuid` in [schema](https://github.com/apache/hudi/blob/master/hudi-spark/src/main/java/org/apache/hudi/QuickstartUtils.java#L58)), partition field (`region/country/city`) and combine logic (`ts` in
 [schema](https://github.com/apache/hudi/blob/master/hudi-spark/src/main/java/org/apache/hudi/QuickstartUtils.java#L58)) to ensure trip records are unique within each partition. For more info, refer to
 [Modeling data stored in Hudi](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=113709185#FAQ-HowdoImodelthedatastoredinHudi)
-and for info on ways to ingest data into Hudi, refer to [Writing Hudi Tables](/docs/writing_data).
+and for info on ways to ingest data into Hudi, refer to [Writing Hudi Tables](/docs/next/writing_data).
 Here we are using the default write operation : `upsert`. If you have a workload without updates, you can also issue
-`insert` or `bulk_insert` operations which could be faster. To know more, refer to [Write operations](/docs/write_operations)
+`insert` or `bulk_insert` operations which could be faster. To know more, refer to [Write operations](/docs/next/write_operations)
 :::
+
 </TabItem>
 
 <TabItem value="sparksql">
 
+Users can use 'INSERT_INTO' to insert data into a hudi table using spark-sql. 

Review Comment:
   nit: INSERT INTO



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
+
+### Table Type
+Here is an example of creating a COW table.
+
+```sql
+-- create a non-primary key (or key less) table
+create table if not exists hudi_table2(
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow'
+);
+```
+
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 
+the table and hudi will treat them as key less table. Users can still perform Merge Into, updates and deletes based on any random data column. 
+
+### Primary Key
+Here is an example of creating COW table with a primary key 'id'. For mutable datasets, it is recommended to set appropriate primary key. 
+
+```sql
+-- create a managed cow table
+create table if not exists hudi_table0 (
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+);
+```
+
+### PreCombineField
+Here is an example of creating an MOR external table. The **preCombineField** option
+is used to specify the preCombine field for merge. Generally 'event time' or some other similar column will be used for 
+ordering purpose. Hudi will be able to handle out of order data using the precombine field value.
+
+```sql
+-- create an external mor table
+create table if not exists hudi_table1 (
+  id int, 
+  name string, 
+  price double,
+  ts bigint
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id,name',
+  preCombineField = 'ts' 
+);
+```
+
+### Partitioned Table
+Here is an example of creating a COW partitioned key less table.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+ ) 
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table
+Here is an example of creating a COW un-partitioned table.
+
+```sql
+-- create a cow table, with primaryKey 'uuid' and unpartitioned. 

Review Comment:
   nit: -- create a non-partitioned cow table with primaryKey 'uuid'



##########
website/docs/quick-start-guide.md:
##########
@@ -866,6 +817,12 @@ when not matched then
 
 ```
 
+:::note
+For a Primary keyed Hudi table, the join condition in MIT is expected to contain the primary keys of the table. 

Review Comment:
   what is MIT?



##########
website/docs/quick-start-guide.md:
##########
@@ -557,19 +597,28 @@ select 1 as id, 'a1' as name, 1000 as ts, '2021-12-09' as dt, '10' as hh;
 insert into hudi_cow_pt_tbl partition(dt = '2021-12-09', hh='11') select 2, 'a2', 1000;
 ```
 
-**NOTICE**
-- By default,  if `preCombineKey `  is provided,  `insert into` use `upsert` as the type of write operation, otherwise use `insert`.
-- We support to use `bulk_insert` as the type of write operation, just need to set two configs: `hoodie.sql.bulk.insert.enable` and `hoodie.sql.insert.mode`. Example as follow: 
+:::note
+- `hoodie.spark.sql.insert.into.operation` will determine how records ingested via spark-sql INSERT_INTO will be treated. Possible values are "bulk_insert", "insert"

Review Comment:
   nit: INSERT INTO



##########
website/docs/quick-start-guide.md:
##########
@@ -1538,174 +998,205 @@ values={[
 <TabItem value="scala">
 
 ```scala
-// spark-shell
-spark.
-  read.format("hudi").
-  load(basePath).
-  select("uuid","partitionpath").
-  sort("partitionpath","uuid").
-  show(100, false)
+spark.read.
+  format("hudi").
+  option("as.of.instant", "20210728141108100").
+  load(basePath)
 
-val inserts = convertToStringList(dataGen.generateInserts(10))
-val df = spark.
-  read.json(spark.sparkContext.parallelize(inserts, 2)).
-  filter("partitionpath = 'americas/united_states/san_francisco'")
-df.write.format("hudi").
-  options(getQuickstartWriteConfigs).
-  option(OPERATION.key(),"insert_overwrite").
-  option(PRECOMBINE_FIELD.key(), "ts").
-  option(RECORDKEY_FIELD.key(), "uuid").
-  option(PARTITIONPATH_FIELD.key(), "partitionpath").
-  option(TBL_NAME.key(), tableName).
-  mode(Append).
-  save(basePath)
+spark.read.
+  format("hudi").
+  option("as.of.instant", "2021-07-28 14:11:08.200").
+  load(basePath)
+
+// It is equal to "as.of.instant = 2021-07-28 00:00:00"
+spark.read.
+  format("hudi").
+  option("as.of.instant", "2021-07-28").
+  load(basePath)
 
-// Should have different keys now for San Francisco alone, from query before.
-spark.
-  read.format("hudi").
-  load(basePath).
-  select("uuid","partitionpath").
-  sort("partitionpath","uuid").
-  show(100, false)
 ```
-</TabItem>
 
+</TabItem>
 <TabItem value="python">
 
 ```python
-# pyspark
-spark.read.format("hudi"). \
-    load(basePath). \
-    select(["uuid", "partitionpath"]). \
-    sort(["partitionpath", "uuid"]). \
-    show(n=100, truncate=False)
-    
-inserts = sc._jvm.org.apache.hudi.QuickstartUtils.convertToStringList(dataGen.generateInserts(10)) 
-df = spark.read.json(spark.sparkContext.parallelize(inserts, 2)). \
-    filter("partitionpath = 'americas/united_states/san_francisco'")
-hudi_insert_overwrite_options = {
-    'hoodie.table.name': tableName,
-    'hoodie.datasource.write.recordkey.field': 'uuid',
-    'hoodie.datasource.write.partitionpath.field': 'partitionpath',
-    'hoodie.datasource.write.table.name': tableName,
-    'hoodie.datasource.write.operation': 'insert_overwrite',
-    'hoodie.datasource.write.precombine.field': 'ts',
-    'hoodie.upsert.shuffle.parallelism': 2,
-    'hoodie.insert.shuffle.parallelism': 2
-}
-df.write.format("hudi").options(**hudi_insert_overwrite_options).mode("append").save(basePath)
-spark.read.format("hudi"). \
-    load(basePath). \
-    select(["uuid", "partitionpath"]). \
-    sort(["partitionpath", "uuid"]). \
-    show(n=100, truncate=False)
+#pyspark
+spark.read. \
+  format("hudi"). \
+  option("as.of.instant", "20210728141108"). \
+  load(basePath)
+
+spark.read. \
+  format("hudi"). \
+  option("as.of.instant", "2021-07-28 14:11:08.000"). \
+  load(basePath)
+
+# It is equal to "as.of.instant = 2021-07-28 00:00:00"
+spark.read. \
+  format("hudi"). \
+  option("as.of.instant", "2021-07-28"). \
+  load(basePath)
 ```
-</TabItem>
 
+</TabItem>
 <TabItem value="sparksql">
 
-`insert overwrite` a partitioned table use the `INSERT_OVERWRITE` type of write operation, while a non-partitioned table to `INSERT_OVERWRITE_TABLE`.
+:::note
+Requires Spark 3.2+
+:::
 
 ```sql
--- insert overwrite non-partitioned table
-insert overwrite hudi_mor_tbl select 99, 'a99', 20.0, 900;
-insert overwrite hudi_cow_nonpcf_tbl select 99, 'a99', 20.0;
 
--- insert overwrite partitioned table with dynamic partition
-insert overwrite table hudi_cow_pt_tbl select 10, 'a10', 1100, '2021-12-09', '10';
-
--- insert overwrite partitioned table with static partition
-insert overwrite hudi_cow_pt_tbl partition(dt = '2021-12-09', hh='12') select 13, 'a13', 1100;
+-- time travel based on commit time, for eg: `20220307091628793`
+select * from hudi_cow_pt_tbl timestamp as of '20220307091628793' where id = 1;
+-- time travel based on different timestamp formats
+select * from hudi_cow_pt_tbl timestamp as of '2022-03-07 09:16:28.100' where id = 1;
+select * from hudi_cow_pt_tbl timestamp as of '2022-03-08' where id = 1;
 ```
+
 </TabItem>
 
 </Tabs
 >
 
-## More Spark SQL Commands
 
-### Alter Table
+### Incremental query
 
-Schema evolution can be achieved via `ALTER TABLE` commands. Below shows some basic examples.
+Hudi also provides capability to obtain a stream of records that changed since given commit timestamp. 
+This can be achieved using Hudi's incremental querying and providing a begin time from which changes need to be streamed. 
+We do not need to specify endTime, if we want all changes after the given commit (as is the common case). 
 
-:::note
-For more detailed examples, please prefer to [schema evolution](schema_evolution)
-:::
+<Tabs
+groupId="programming-language"
+defaultValue="scala"
+values={[
+{ label: 'Scala', value: 'scala', },
+{ label: 'Python', value: 'python', },
+{ label: 'Spark SQL', value: 'sparksql', }
+]}
+>
 
-**Syntax**
-```sql
--- Alter table name
-ALTER TABLE oldTableName RENAME TO newTableName
+<TabItem value="scala">
+
+```scala
+// spark-shell
+// reload data
+spark.
+  read.
+  format("hudi").
+  load(basePath).
+  createOrReplaceTempView("hudi_trips_snapshot")
 
--- Alter table add columns
-ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+val commits = spark.sql("select distinct(_hoodie_commit_time) as commitTime from  hudi_trips_snapshot order by commitTime").map(k => k.getString(0)).take(50)
+val beginTime = commits(commits.length - 2) // commit time we are interested in
 
--- Alter table column type
-ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+// incrementally query data
+val tripsIncrementalDF = spark.read.format("hudi").
+  option(QUERY_TYPE.key(), QUERY_TYPE_INCREMENTAL_OPT_VAL).
+  option(BEGIN_INSTANTTIME.key(), beginTime).
+  load(basePath)
+tripsIncrementalDF.createOrReplaceTempView("hudi_trips_incremental")
 
--- Alter table properties
-ALTER TABLE tableIdentifier SET TBLPROPERTIES (key = 'value')
+spark.sql("select `_hoodie_commit_time`, fare, begin_lon, begin_lat, ts from  hudi_trips_incremental where fare > 20.0").show()
 ```
-**Examples**
-```sql
---rename to:
-ALTER TABLE hudi_cow_nonpcf_tbl RENAME TO hudi_cow_nonpcf_tbl2;
 
---add column:
-ALTER TABLE hudi_cow_nonpcf_tbl2 add columns(remark string);
+</TabItem>
+<TabItem value="python">
+
+```python
+# pyspark
+# reload data
+spark. \
+  read. \
+  format("hudi"). \
+  load(basePath). \
+  createOrReplaceTempView("hudi_trips_snapshot")
+
+commits = list(map(lambda row: row[0], spark.sql("select distinct(_hoodie_commit_time) as commitTime from  hudi_trips_snapshot order by commitTime").limit(50).collect()))
+beginTime = commits[len(commits) - 2] # commit time we are interested in
+
+# incrementally query data
+incremental_read_options = {
+  'hoodie.datasource.query.type': 'incremental',
+  'hoodie.datasource.read.begin.instanttime': beginTime,
+}
 
---change column:
-ALTER TABLE hudi_cow_nonpcf_tbl2 change column uuid uuid bigint;
+tripsIncrementalDF = spark.read.format("hudi"). \
+  options(**incremental_read_options). \
+  load(basePath)
+tripsIncrementalDF.createOrReplaceTempView("hudi_trips_incremental")
 
---set properties;
-alter table hudi_cow_nonpcf_tbl2 set tblproperties (hoodie.keep.max.commits = '10');
+spark.sql("select `_hoodie_commit_time`, fare, begin_lon, begin_lat, ts from  hudi_trips_incremental where fare > 20.0").show()
 ```
 
-### Partition SQL Command
+</TabItem>
 
-**Syntax**
+<TabItem value="sparksql">
 
 ```sql
--- Add Partition
-ALTER TABLE tableIdentifier ADD PARTITION ( partition_col_name = partition_col_val [ , ... ] )
+-- syntax
+hudi_table_changes(table or path, queryType, beginTime [, endTime]);  
+-- table or path: table identifier, example: db.tableName, tableName, 
+--                or path for of your table, example: path/to/hudiTable  
+--                in this case table does not need to exist in the metastore,
+-- queryType: incremental query mode, example: latest_state, cdc  
+--            (for cdc query, first enable cdc for your table by setting cdc.enabled=true),
+-- beginTime: instantTime to begin query from, example: earliest, 202305150000, 
+-- endTime: optional instantTime to end query at, example: 202305160000, 
 
--- Drop Partition
-ALTER TABLE tableIdentifier DROP PARTITION ( partition_col_name = partition_col_val [ , ... ] )
+-- incrementally query data by table name
+-- start from earliest available commit, end at latest available commit.  
+select * from hudi_table_changes('db.table', 'latest_state', 'earliest');
 
--- Show Partitions
-SHOW PARTITIONS tableIdentifier
-```
-**Examples**
-```sql
---show partition:
-show partitions hudi_cow_pt_tbl;
+-- start from earliest, end at 202305160000.  
+select * from hudi_table_changes('table', 'latest_state', 'earliest', '202305160000');  
+
+-- start from 202305150000, end at 202305160000.
+select * from hudi_table_changes('table', 'latest_state', '202305150000', '202305160000');
+
+-- incrementally query data by path
+-- start from earliest available commit, end at latest available commit.
+select * from hudi_table_changes('path/to/table', 'cdc', 'earliest');
+
+-- start from earliest, end at 202305160000.
+select * from hudi_table_changes('path/to/table', 'cdc', 'earliest', '202305160000');
+
+-- start from 202305150000, end at 202305160000.
+select * from hudi_table_changes('path/to/table', 'cdc', '202305150000', '202305160000');
 
---drop partition：
-alter table hudi_cow_pt_tbl drop partition (dt='2021-12-09', hh='10');
 ```
-:::note
-Currently,  the result of `show partitions` is based on the filesystem table path. It's not precise when delete the whole partition data or drop certain partition directly.
 
+</TabItem>
+
+</Tabs
+>
+
+:::info
+This will give all changes that happened after the beginTime commit with the filter of fare > 20.0. The unique thing about this
+feature is that it now lets you author streaming pipelines on batch data.
 :::
 
-### Procedures
+## Streaming writers
+### Hudi Streamer
+Hudi provides a ingestion tool to assist with ingesting data into hudi from various difference sources in a streaming manner. 
+This has lot of niceties like auto checkpointing, schema enforcement via schema provider, transformation support and so on.
+Please refer to [here](/docs/next/hoodie_deltastreamer#hudi-streamer) for more info. 
 
-**Syntax**
-```sql
---Call procedure by positional arguments
-CALL system.procedure_name(arg_1, arg_2, ... arg_n)
+### Structured Streaming
 
---Call procedure by named arguments
-CALL system.procedure_name(arg_name_2 => arg_2, arg_name_1 => arg_1, ... arg_name_n => arg_n)
-```
-**Examples**
-```sql
---show commit's info
-call show_commits(table => 'test_hudi_table', limit => 10);
-```
+Hudi supports Spark Structured Streaming reads and writes as well. Please refer to [here](/docs/next/toBeFixed) for more info.

Review Comment:
   the link needs to be fixed



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
+
+### Table Type
+Here is an example of creating a COW table.
+
+```sql
+-- create a non-primary key (or key less) table
+create table if not exists hudi_table2(
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow'
+);
+```
+
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 
+the table and hudi will treat them as key less table. Users can still perform Merge Into, updates and deletes based on any random data column. 
+
+### Primary Key
+Here is an example of creating COW table with a primary key 'id'. For mutable datasets, it is recommended to set appropriate primary key. 
+
+```sql
+-- create a managed cow table
+create table if not exists hudi_table0 (
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+);
+```
+
+### PreCombineField
+Here is an example of creating an MOR external table. The **preCombineField** option
+is used to specify the preCombine field for merge. Generally 'event time' or some other similar column will be used for 
+ordering purpose. Hudi will be able to handle out of order data using the precombine field value.
+
+```sql
+-- create an external mor table
+create table if not exists hudi_table1 (
+  id int, 
+  name string, 
+  price double,
+  ts bigint
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id,name',
+  preCombineField = 'ts' 
+);
+```
+
+### Partitioned Table
+Here is an example of creating a COW partitioned key less table.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+ ) 
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table
+Here is an example of creating a COW un-partitioned table.
+
+```sql
+-- create a cow table, with primaryKey 'uuid' and unpartitioned. 
+create table hudi_cow_nonpcf_tbl (
+  uuid int,
+  name string,
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'uuid'
+);
+```
+
+Here is an example of creating a MOR un-partitioned table.
+
+```sql
+-- create a mor non-partitioned table with preCombineField provided
+create table hudi_mor_tbl (
+    id int,
+    name string,
+    price double,
+    ts bigint
+) using hudi
+tblproperties (
+    type = 'mor',
+    primaryKey = 'id',
+    preCombineField = 'ts'
+);
+```
+
+### Create Table for an External Hudi Table
+You can create an External table using the `location` statement. If an external location is not specified it is considered a managed table. 
+You can read more about external vs managed tables [here](https://sparkbyexamples.com/apache-hive/difference-between-hive-internal-tables-and-external-tables/).
+An external table is useful if you need to read/write to/from a pre-existing hudi table.
+
+```sql
+ create table h_p1 using hudi
+ location '/path/to/hudi';
+```
+
+:::tip
+You don't need to specify schema and any properties except the partitioned columns if existed. Hudi can automatically recognize the schema and configurations.
+:::
+
+### Create Table AS SELECT
+
+Hudi supports CTAS(Create table as select) on spark sql. <br/>
+**Note:** For better performance to load data to hudi table, CTAS uses **bulk insert** as the write operation.
+
+**Example CTAS command to create a non-partitioned COW key less table.**
+
+```sql 
+create table h3 using hudi
+tblproperties (type = 'cow')
+as
+select 1 as id, 'a1' as name, 10 as price;
+```
+
+**Example CTAS command to create a partitioned, primary keyed COW table.**
+
+```sql
+create table h2 using hudi
+tblproperties (type = 'cow', primaryKey = 'id')
+partitioned by (dt)
+as
+select 1 as id, 'a1' as name, 10 as price, 1000 as dt;
+```
+
+**Example CTAS command to load data from another table.**
+
+```sql
+# create managed parquet table 
+create table parquet_mngd using parquet location 'file:///tmp/parquet_dataset/*.parquet';
+
+# CTAS by loading data into hudi table
+create table hudi_tbl using hudi location 'file:/tmp/hudi/hudi_tbl/' tblproperties ( 
+  type = 'cow', 
+  primaryKey = 'id', 
+  preCombineField = 'ts' 
+ ) 
+partitioned by (datestr) as select * from parquet_mngd;
+```
+
+### Set hoodie config options
+You can also set the config with table options when creating table which will work for
+the table scope only and override the config set by the SET command.
+```sql
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  ${hoodie.config.key1} = '${hoodie.config.value2}',
+  ${hoodie.config.key2} = '${hoodie.config.value2}',
+  ....
+);
+
+e.g.
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  hoodie.cleaner.fileversions.retained = '20',
+  hoodie.keep.max.commits = '20'
+);
+```
+
+## Spark Alter Table
+### Syntax
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+```
+
+:::note
+`ALTER TABLE ... RENAME TO ...` is not supported when using AWS Glue Data Catalog as hive metastore as Glue itself does 
+not support table renames.
+:::
+
+### Examples
+```sql
+alter table h0 rename to h0_1;
+
+alter table h0_1 add columns(ext0 string);
+
+alter table h0_1 change column id id bigint;
+```
+
+Schema evolution can be achieved via `ALTER TABLE` commands. Below shows some basic examples.
+
+:::note
+For more detailed examples, please prefer to [schema evolution](/docs/schema_evolution)
+:::
+
+**Syntax**
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)

Review Comment:
   nit: (, colAndType) -> [, colAndType]



##########
website/docs/quick-start-guide.md:
##########
@@ -1538,174 +998,205 @@ values={[
 <TabItem value="scala">
 
 ```scala
-// spark-shell
-spark.
-  read.format("hudi").
-  load(basePath).
-  select("uuid","partitionpath").
-  sort("partitionpath","uuid").
-  show(100, false)
+spark.read.
+  format("hudi").
+  option("as.of.instant", "20210728141108100").
+  load(basePath)
 
-val inserts = convertToStringList(dataGen.generateInserts(10))
-val df = spark.
-  read.json(spark.sparkContext.parallelize(inserts, 2)).
-  filter("partitionpath = 'americas/united_states/san_francisco'")
-df.write.format("hudi").
-  options(getQuickstartWriteConfigs).
-  option(OPERATION.key(),"insert_overwrite").
-  option(PRECOMBINE_FIELD.key(), "ts").
-  option(RECORDKEY_FIELD.key(), "uuid").
-  option(PARTITIONPATH_FIELD.key(), "partitionpath").
-  option(TBL_NAME.key(), tableName).
-  mode(Append).
-  save(basePath)
+spark.read.
+  format("hudi").
+  option("as.of.instant", "2021-07-28 14:11:08.200").
+  load(basePath)
+
+// It is equal to "as.of.instant = 2021-07-28 00:00:00"
+spark.read.
+  format("hudi").
+  option("as.of.instant", "2021-07-28").
+  load(basePath)
 
-// Should have different keys now for San Francisco alone, from query before.
-spark.
-  read.format("hudi").
-  load(basePath).
-  select("uuid","partitionpath").
-  sort("partitionpath","uuid").
-  show(100, false)
 ```
-</TabItem>
 
+</TabItem>
 <TabItem value="python">
 
 ```python
-# pyspark
-spark.read.format("hudi"). \
-    load(basePath). \
-    select(["uuid", "partitionpath"]). \
-    sort(["partitionpath", "uuid"]). \
-    show(n=100, truncate=False)
-    
-inserts = sc._jvm.org.apache.hudi.QuickstartUtils.convertToStringList(dataGen.generateInserts(10)) 
-df = spark.read.json(spark.sparkContext.parallelize(inserts, 2)). \
-    filter("partitionpath = 'americas/united_states/san_francisco'")
-hudi_insert_overwrite_options = {
-    'hoodie.table.name': tableName,
-    'hoodie.datasource.write.recordkey.field': 'uuid',
-    'hoodie.datasource.write.partitionpath.field': 'partitionpath',
-    'hoodie.datasource.write.table.name': tableName,
-    'hoodie.datasource.write.operation': 'insert_overwrite',
-    'hoodie.datasource.write.precombine.field': 'ts',
-    'hoodie.upsert.shuffle.parallelism': 2,
-    'hoodie.insert.shuffle.parallelism': 2
-}
-df.write.format("hudi").options(**hudi_insert_overwrite_options).mode("append").save(basePath)
-spark.read.format("hudi"). \
-    load(basePath). \
-    select(["uuid", "partitionpath"]). \
-    sort(["partitionpath", "uuid"]). \
-    show(n=100, truncate=False)
+#pyspark
+spark.read. \
+  format("hudi"). \
+  option("as.of.instant", "20210728141108"). \
+  load(basePath)
+
+spark.read. \
+  format("hudi"). \
+  option("as.of.instant", "2021-07-28 14:11:08.000"). \
+  load(basePath)
+
+# It is equal to "as.of.instant = 2021-07-28 00:00:00"
+spark.read. \
+  format("hudi"). \
+  option("as.of.instant", "2021-07-28"). \
+  load(basePath)
 ```
-</TabItem>
 
+</TabItem>
 <TabItem value="sparksql">
 
-`insert overwrite` a partitioned table use the `INSERT_OVERWRITE` type of write operation, while a non-partitioned table to `INSERT_OVERWRITE_TABLE`.
+:::note
+Requires Spark 3.2+
+:::
 
 ```sql
--- insert overwrite non-partitioned table
-insert overwrite hudi_mor_tbl select 99, 'a99', 20.0, 900;
-insert overwrite hudi_cow_nonpcf_tbl select 99, 'a99', 20.0;
 
--- insert overwrite partitioned table with dynamic partition
-insert overwrite table hudi_cow_pt_tbl select 10, 'a10', 1100, '2021-12-09', '10';
-
--- insert overwrite partitioned table with static partition
-insert overwrite hudi_cow_pt_tbl partition(dt = '2021-12-09', hh='12') select 13, 'a13', 1100;
+-- time travel based on commit time, for eg: `20220307091628793`
+select * from hudi_cow_pt_tbl timestamp as of '20220307091628793' where id = 1;
+-- time travel based on different timestamp formats
+select * from hudi_cow_pt_tbl timestamp as of '2022-03-07 09:16:28.100' where id = 1;
+select * from hudi_cow_pt_tbl timestamp as of '2022-03-08' where id = 1;
 ```
+
 </TabItem>
 
 </Tabs
 >
 
-## More Spark SQL Commands
 
-### Alter Table
+### Incremental query
 
-Schema evolution can be achieved via `ALTER TABLE` commands. Below shows some basic examples.
+Hudi also provides capability to obtain a stream of records that changed since given commit timestamp. 
+This can be achieved using Hudi's incremental querying and providing a begin time from which changes need to be streamed. 

Review Comment:
   nit: querying and providing -> querying by providing



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
+
+### Table Type
+Here is an example of creating a COW table.
+
+```sql
+-- create a non-primary key (or key less) table
+create table if not exists hudi_table2(
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow'
+);
+```
+
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 
+the table and hudi will treat them as key less table. Users can still perform Merge Into, updates and deletes based on any random data column. 
+
+### Primary Key
+Here is an example of creating COW table with a primary key 'id'. For mutable datasets, it is recommended to set appropriate primary key. 
+
+```sql
+-- create a managed cow table
+create table if not exists hudi_table0 (
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+);
+```
+
+### PreCombineField
+Here is an example of creating an MOR external table. The **preCombineField** option
+is used to specify the preCombine field for merge. Generally 'event time' or some other similar column will be used for 
+ordering purpose. Hudi will be able to handle out of order data using the precombine field value.
+
+```sql
+-- create an external mor table
+create table if not exists hudi_table1 (
+  id int, 
+  name string, 
+  price double,
+  ts bigint
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id,name',
+  preCombineField = 'ts' 
+);
+```
+
+### Partitioned Table
+Here is an example of creating a COW partitioned key less table.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+ ) 
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table
+Here is an example of creating a COW un-partitioned table.
+
+```sql
+-- create a cow table, with primaryKey 'uuid' and unpartitioned. 
+create table hudi_cow_nonpcf_tbl (
+  uuid int,
+  name string,
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'uuid'
+);
+```
+
+Here is an example of creating a MOR un-partitioned table.
+
+```sql
+-- create a mor non-partitioned table with preCombineField provided
+create table hudi_mor_tbl (
+    id int,
+    name string,
+    price double,
+    ts bigint
+) using hudi
+tblproperties (
+    type = 'mor',
+    primaryKey = 'id',
+    preCombineField = 'ts'
+);
+```
+
+### Create Table for an External Hudi Table
+You can create an External table using the `location` statement. If an external location is not specified it is considered a managed table. 
+You can read more about external vs managed tables [here](https://sparkbyexamples.com/apache-hive/difference-between-hive-internal-tables-and-external-tables/).
+An external table is useful if you need to read/write to/from a pre-existing hudi table.
+
+```sql
+ create table h_p1 using hudi
+ location '/path/to/hudi';
+```
+
+:::tip
+You don't need to specify schema and any properties except the partitioned columns if existed. Hudi can automatically recognize the schema and configurations.
+:::
+
+### Create Table AS SELECT
+
+Hudi supports CTAS(Create table as select) on spark sql. <br/>
+**Note:** For better performance to load data to hudi table, CTAS uses **bulk insert** as the write operation.
+
+**Example CTAS command to create a non-partitioned COW key less table.**
+
+```sql 
+create table h3 using hudi
+tblproperties (type = 'cow')
+as
+select 1 as id, 'a1' as name, 10 as price;
+```
+
+**Example CTAS command to create a partitioned, primary keyed COW table.**
+
+```sql
+create table h2 using hudi
+tblproperties (type = 'cow', primaryKey = 'id')
+partitioned by (dt)
+as
+select 1 as id, 'a1' as name, 10 as price, 1000 as dt;
+```
+
+**Example CTAS command to load data from another table.**
+
+```sql
+# create managed parquet table 
+create table parquet_mngd using parquet location 'file:///tmp/parquet_dataset/*.parquet';
+
+# CTAS by loading data into hudi table
+create table hudi_tbl using hudi location 'file:/tmp/hudi/hudi_tbl/' tblproperties ( 
+  type = 'cow', 
+  primaryKey = 'id', 
+  preCombineField = 'ts' 
+ ) 
+partitioned by (datestr) as select * from parquet_mngd;
+```
+
+### Set hoodie config options
+You can also set the config with table options when creating table which will work for
+the table scope only and override the config set by the SET command.
+```sql
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  ${hoodie.config.key1} = '${hoodie.config.value2}',
+  ${hoodie.config.key2} = '${hoodie.config.value2}',
+  ....
+);
+
+e.g.
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  hoodie.cleaner.fileversions.retained = '20',
+  hoodie.keep.max.commits = '20'
+);
+```
+
+## Spark Alter Table
+### Syntax
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+```
+
+:::note
+`ALTER TABLE ... RENAME TO ...` is not supported when using AWS Glue Data Catalog as hive metastore as Glue itself does 
+not support table renames.
+:::
+
+### Examples
+```sql
+alter table h0 rename to h0_1;
+
+alter table h0_1 add columns(ext0 string);
+
+alter table h0_1 change column id id bigint;
+```
+
+Schema evolution can be achieved via `ALTER TABLE` commands. Below shows some basic examples.
+
+:::note
+For more detailed examples, please prefer to [schema evolution](/docs/schema_evolution)
+:::
+
+**Syntax**
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+
+-- Alter table properties
+ALTER TABLE tableIdentifier SET TBLPROPERTIES (key = 'value')
+```
+
+**Examples**
+```sql
+--rename to:
+ALTER TABLE hudi_cow_nonpcf_tbl RENAME TO hudi_cow_nonpcf_tbl2;
+
+--add column:
+ALTER TABLE hudi_cow_nonpcf_tbl2 add columns(remark string);
+
+--change column:
+ALTER TABLE hudi_cow_nonpcf_tbl2 change column uuid uuid bigint;
+
+--set properties;
+alter table hudi_cow_nonpcf_tbl2 set tblproperties (hoodie.keep.max.commits = '10');
+```
+
+### Alter hoodie config options
+You can also alter the write config for a table by the **ALTER SERDEPROPERTIES**
+
+Example:
+```sql
+ alter table h3 set serdeproperties (hoodie.keep.max.commits = '10') 
+```
+
+### Use set command
+You can use the **set** command to set any custom hudi's config, which will work for the
+whole spark session scope.
+```sql
+set hoodie.insert.shuffle.parallelism = 100;
+set hoodie.upsert.shuffle.parallelism = 100;
+set hoodie.delete.shuffle.parallelism = 100;
+```
+
+### Show and Drop partitions
+
+**Syntax**
+
+```sql
+-- Show Partitions
+SHOW PARTITIONS tableIdentifier
+
+-- Drop Partition
+ALTER TABLE tableIdentifier DROP PARTITION ( partition_col_name = partition_col_val [ , ... ] )
+```
+
+**Examples**
+```sql
+--show partition:
+show partitions hudi_cow_pt_tbl;
+
+--drop partition：
+alter table hudi_cow_pt_tbl drop partition (dt='2021-12-09', hh='10');
+```
+
+:::note
+Currently,  the result of `show partitions` is based on the filesystem table path. It's not precise when delete the whole partition data or drop certain partition directly.
+:::
+
+## Flink
+
+### Create Catalog
+
+The catalog helps to manage the SQL tables, the table can be shared among CLI sessions if the catalog persists the table DDLs.
+For `hms` mode, the catalog also supplements the hive syncing options.
+
+HMS mode catalog SQL demo:
+```sql
+CREATE CATALOG hoodie_catalog
+  WITH (
+    'type'='hudi',
+    'catalog.path' = '${catalog default root path}',
+    'hive.conf.dir' = '${directory where hive-site.xml is located}',
+    'mode'='hms' -- supports 'dfs' mode that uses the DFS backend for table DDLs persistence
+  );
+```
+
+#### Options
+|  Option Name  | Required | Default | Remarks |
+|  -----------  | -------  | ------- | ------- |
+| `catalog.path` | true | -- | Default root path for the catalog, the path is used to infer the table path automatically, the default table path: `${catalog.path}/${db_name}/${table_name}` |
+| `default-database` | false | default | default database name |
+| `hive.conf.dir` | false | -- | The directory where hive-site.xml is located, only valid in `hms` mode |
+| `mode` | false | dfs | Supports `hms` mode that uses HMS to persist the table options |

Review Comment:
   Let us mention all the different values possible in remarks



##########
website/docs/quick-start-guide.md:
##########
@@ -1538,174 +998,205 @@ values={[
 <TabItem value="scala">
 
 ```scala
-// spark-shell
-spark.
-  read.format("hudi").
-  load(basePath).
-  select("uuid","partitionpath").
-  sort("partitionpath","uuid").
-  show(100, false)
+spark.read.
+  format("hudi").
+  option("as.of.instant", "20210728141108100").
+  load(basePath)
 
-val inserts = convertToStringList(dataGen.generateInserts(10))
-val df = spark.
-  read.json(spark.sparkContext.parallelize(inserts, 2)).
-  filter("partitionpath = 'americas/united_states/san_francisco'")
-df.write.format("hudi").
-  options(getQuickstartWriteConfigs).
-  option(OPERATION.key(),"insert_overwrite").
-  option(PRECOMBINE_FIELD.key(), "ts").
-  option(RECORDKEY_FIELD.key(), "uuid").
-  option(PARTITIONPATH_FIELD.key(), "partitionpath").
-  option(TBL_NAME.key(), tableName).
-  mode(Append).
-  save(basePath)
+spark.read.
+  format("hudi").
+  option("as.of.instant", "2021-07-28 14:11:08.200").
+  load(basePath)
+
+// It is equal to "as.of.instant = 2021-07-28 00:00:00"
+spark.read.
+  format("hudi").
+  option("as.of.instant", "2021-07-28").
+  load(basePath)
 
-// Should have different keys now for San Francisco alone, from query before.
-spark.
-  read.format("hudi").
-  load(basePath).
-  select("uuid","partitionpath").
-  sort("partitionpath","uuid").
-  show(100, false)
 ```
-</TabItem>
 
+</TabItem>
 <TabItem value="python">
 
 ```python
-# pyspark
-spark.read.format("hudi"). \
-    load(basePath). \
-    select(["uuid", "partitionpath"]). \
-    sort(["partitionpath", "uuid"]). \
-    show(n=100, truncate=False)
-    
-inserts = sc._jvm.org.apache.hudi.QuickstartUtils.convertToStringList(dataGen.generateInserts(10)) 
-df = spark.read.json(spark.sparkContext.parallelize(inserts, 2)). \
-    filter("partitionpath = 'americas/united_states/san_francisco'")
-hudi_insert_overwrite_options = {
-    'hoodie.table.name': tableName,
-    'hoodie.datasource.write.recordkey.field': 'uuid',
-    'hoodie.datasource.write.partitionpath.field': 'partitionpath',
-    'hoodie.datasource.write.table.name': tableName,
-    'hoodie.datasource.write.operation': 'insert_overwrite',
-    'hoodie.datasource.write.precombine.field': 'ts',
-    'hoodie.upsert.shuffle.parallelism': 2,
-    'hoodie.insert.shuffle.parallelism': 2
-}
-df.write.format("hudi").options(**hudi_insert_overwrite_options).mode("append").save(basePath)
-spark.read.format("hudi"). \
-    load(basePath). \
-    select(["uuid", "partitionpath"]). \
-    sort(["partitionpath", "uuid"]). \
-    show(n=100, truncate=False)
+#pyspark
+spark.read. \
+  format("hudi"). \
+  option("as.of.instant", "20210728141108"). \
+  load(basePath)
+
+spark.read. \
+  format("hudi"). \
+  option("as.of.instant", "2021-07-28 14:11:08.000"). \
+  load(basePath)
+
+# It is equal to "as.of.instant = 2021-07-28 00:00:00"
+spark.read. \
+  format("hudi"). \
+  option("as.of.instant", "2021-07-28"). \
+  load(basePath)
 ```
-</TabItem>
 
+</TabItem>
 <TabItem value="sparksql">
 
-`insert overwrite` a partitioned table use the `INSERT_OVERWRITE` type of write operation, while a non-partitioned table to `INSERT_OVERWRITE_TABLE`.
+:::note
+Requires Spark 3.2+
+:::
 
 ```sql
--- insert overwrite non-partitioned table
-insert overwrite hudi_mor_tbl select 99, 'a99', 20.0, 900;
-insert overwrite hudi_cow_nonpcf_tbl select 99, 'a99', 20.0;
 
--- insert overwrite partitioned table with dynamic partition
-insert overwrite table hudi_cow_pt_tbl select 10, 'a10', 1100, '2021-12-09', '10';
-
--- insert overwrite partitioned table with static partition
-insert overwrite hudi_cow_pt_tbl partition(dt = '2021-12-09', hh='12') select 13, 'a13', 1100;
+-- time travel based on commit time, for eg: `20220307091628793`
+select * from hudi_cow_pt_tbl timestamp as of '20220307091628793' where id = 1;
+-- time travel based on different timestamp formats
+select * from hudi_cow_pt_tbl timestamp as of '2022-03-07 09:16:28.100' where id = 1;
+select * from hudi_cow_pt_tbl timestamp as of '2022-03-08' where id = 1;
 ```
+
 </TabItem>
 
 </Tabs
 >
 
-## More Spark SQL Commands
 
-### Alter Table
+### Incremental query
 
-Schema evolution can be achieved via `ALTER TABLE` commands. Below shows some basic examples.
+Hudi also provides capability to obtain a stream of records that changed since given commit timestamp. 
+This can be achieved using Hudi's incremental querying and providing a begin time from which changes need to be streamed. 
+We do not need to specify endTime, if we want all changes after the given commit (as is the common case). 
 
-:::note
-For more detailed examples, please prefer to [schema evolution](schema_evolution)
-:::
+<Tabs
+groupId="programming-language"
+defaultValue="scala"
+values={[
+{ label: 'Scala', value: 'scala', },
+{ label: 'Python', value: 'python', },
+{ label: 'Spark SQL', value: 'sparksql', }
+]}
+>
 
-**Syntax**
-```sql
--- Alter table name
-ALTER TABLE oldTableName RENAME TO newTableName
+<TabItem value="scala">
+
+```scala
+// spark-shell
+// reload data
+spark.
+  read.
+  format("hudi").
+  load(basePath).
+  createOrReplaceTempView("hudi_trips_snapshot")
 
--- Alter table add columns
-ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+val commits = spark.sql("select distinct(_hoodie_commit_time) as commitTime from  hudi_trips_snapshot order by commitTime").map(k => k.getString(0)).take(50)
+val beginTime = commits(commits.length - 2) // commit time we are interested in
 
--- Alter table column type
-ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+// incrementally query data
+val tripsIncrementalDF = spark.read.format("hudi").
+  option(QUERY_TYPE.key(), QUERY_TYPE_INCREMENTAL_OPT_VAL).
+  option(BEGIN_INSTANTTIME.key(), beginTime).
+  load(basePath)
+tripsIncrementalDF.createOrReplaceTempView("hudi_trips_incremental")
 
--- Alter table properties
-ALTER TABLE tableIdentifier SET TBLPROPERTIES (key = 'value')
+spark.sql("select `_hoodie_commit_time`, fare, begin_lon, begin_lat, ts from  hudi_trips_incremental where fare > 20.0").show()
 ```
-**Examples**
-```sql
---rename to:
-ALTER TABLE hudi_cow_nonpcf_tbl RENAME TO hudi_cow_nonpcf_tbl2;
 
---add column:
-ALTER TABLE hudi_cow_nonpcf_tbl2 add columns(remark string);
+</TabItem>
+<TabItem value="python">
+
+```python
+# pyspark
+# reload data
+spark. \
+  read. \
+  format("hudi"). \
+  load(basePath). \
+  createOrReplaceTempView("hudi_trips_snapshot")
+
+commits = list(map(lambda row: row[0], spark.sql("select distinct(_hoodie_commit_time) as commitTime from  hudi_trips_snapshot order by commitTime").limit(50).collect()))
+beginTime = commits[len(commits) - 2] # commit time we are interested in
+
+# incrementally query data
+incremental_read_options = {
+  'hoodie.datasource.query.type': 'incremental',
+  'hoodie.datasource.read.begin.instanttime': beginTime,
+}
 
---change column:
-ALTER TABLE hudi_cow_nonpcf_tbl2 change column uuid uuid bigint;
+tripsIncrementalDF = spark.read.format("hudi"). \
+  options(**incremental_read_options). \
+  load(basePath)
+tripsIncrementalDF.createOrReplaceTempView("hudi_trips_incremental")
 
---set properties;
-alter table hudi_cow_nonpcf_tbl2 set tblproperties (hoodie.keep.max.commits = '10');
+spark.sql("select `_hoodie_commit_time`, fare, begin_lon, begin_lat, ts from  hudi_trips_incremental where fare > 20.0").show()
 ```
 
-### Partition SQL Command
+</TabItem>
 
-**Syntax**
+<TabItem value="sparksql">
 
 ```sql
--- Add Partition
-ALTER TABLE tableIdentifier ADD PARTITION ( partition_col_name = partition_col_val [ , ... ] )
+-- syntax
+hudi_table_changes(table or path, queryType, beginTime [, endTime]);  
+-- table or path: table identifier, example: db.tableName, tableName, 
+--                or path for of your table, example: path/to/hudiTable  
+--                in this case table does not need to exist in the metastore,
+-- queryType: incremental query mode, example: latest_state, cdc  
+--            (for cdc query, first enable cdc for your table by setting cdc.enabled=true),
+-- beginTime: instantTime to begin query from, example: earliest, 202305150000, 
+-- endTime: optional instantTime to end query at, example: 202305160000, 
 
--- Drop Partition
-ALTER TABLE tableIdentifier DROP PARTITION ( partition_col_name = partition_col_val [ , ... ] )
+-- incrementally query data by table name
+-- start from earliest available commit, end at latest available commit.  
+select * from hudi_table_changes('db.table', 'latest_state', 'earliest');
 
--- Show Partitions
-SHOW PARTITIONS tableIdentifier
-```
-**Examples**
-```sql
---show partition:
-show partitions hudi_cow_pt_tbl;
+-- start from earliest, end at 202305160000.  
+select * from hudi_table_changes('table', 'latest_state', 'earliest', '202305160000');  
+
+-- start from 202305150000, end at 202305160000.
+select * from hudi_table_changes('table', 'latest_state', '202305150000', '202305160000');
+
+-- incrementally query data by path
+-- start from earliest available commit, end at latest available commit.
+select * from hudi_table_changes('path/to/table', 'cdc', 'earliest');
+
+-- start from earliest, end at 202305160000.
+select * from hudi_table_changes('path/to/table', 'cdc', 'earliest', '202305160000');
+
+-- start from 202305150000, end at 202305160000.
+select * from hudi_table_changes('path/to/table', 'cdc', '202305150000', '202305160000');
 
---drop partition：
-alter table hudi_cow_pt_tbl drop partition (dt='2021-12-09', hh='10');
 ```
-:::note
-Currently,  the result of `show partitions` is based on the filesystem table path. It's not precise when delete the whole partition data or drop certain partition directly.
 
+</TabItem>
+
+</Tabs
+>
+
+:::info
+This will give all changes that happened after the beginTime commit with the filter of fare > 20.0. The unique thing about this
+feature is that it now lets you author streaming pipelines on batch data.
 :::
 
-### Procedures
+## Streaming writers
+### Hudi Streamer
+Hudi provides a ingestion tool to assist with ingesting data into hudi from various difference sources in a streaming manner. 

Review Comment:
   nit: remove difference



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
+
+### Table Type
+Here is an example of creating a COW table.
+
+```sql
+-- create a non-primary key (or key less) table
+create table if not exists hudi_table2(
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow'
+);
+```
+
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 
+the table and hudi will treat them as key less table. Users can still perform Merge Into, updates and deletes based on any random data column. 
+
+### Primary Key
+Here is an example of creating COW table with a primary key 'id'. For mutable datasets, it is recommended to set appropriate primary key. 
+
+```sql
+-- create a managed cow table
+create table if not exists hudi_table0 (
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+);
+```
+
+### PreCombineField
+Here is an example of creating an MOR external table. The **preCombineField** option
+is used to specify the preCombine field for merge. Generally 'event time' or some other similar column will be used for 
+ordering purpose. Hudi will be able to handle out of order data using the precombine field value.
+
+```sql
+-- create an external mor table
+create table if not exists hudi_table1 (
+  id int, 
+  name string, 
+  price double,
+  ts bigint
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id,name',
+  preCombineField = 'ts' 
+);
+```
+
+### Partitioned Table
+Here is an example of creating a COW partitioned key less table.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+ ) 
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table
+Here is an example of creating a COW un-partitioned table.
+
+```sql
+-- create a cow table, with primaryKey 'uuid' and unpartitioned. 
+create table hudi_cow_nonpcf_tbl (
+  uuid int,
+  name string,
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'uuid'
+);
+```
+
+Here is an example of creating a MOR un-partitioned table.
+
+```sql
+-- create a mor non-partitioned table with preCombineField provided
+create table hudi_mor_tbl (
+    id int,
+    name string,
+    price double,
+    ts bigint
+) using hudi
+tblproperties (
+    type = 'mor',
+    primaryKey = 'id',
+    preCombineField = 'ts'
+);
+```
+
+### Create Table for an External Hudi Table
+You can create an External table using the `location` statement. If an external location is not specified it is considered a managed table. 
+You can read more about external vs managed tables [here](https://sparkbyexamples.com/apache-hive/difference-between-hive-internal-tables-and-external-tables/).
+An external table is useful if you need to read/write to/from a pre-existing hudi table.
+
+```sql
+ create table h_p1 using hudi
+ location '/path/to/hudi';
+```
+
+:::tip
+You don't need to specify schema and any properties except the partitioned columns if existed. Hudi can automatically recognize the schema and configurations.
+:::
+
+### Create Table AS SELECT
+
+Hudi supports CTAS(Create table as select) on spark sql. <br/>
+**Note:** For better performance to load data to hudi table, CTAS uses **bulk insert** as the write operation.
+
+**Example CTAS command to create a non-partitioned COW key less table.**
+
+```sql 
+create table h3 using hudi
+tblproperties (type = 'cow')
+as
+select 1 as id, 'a1' as name, 10 as price;
+```
+
+**Example CTAS command to create a partitioned, primary keyed COW table.**
+
+```sql
+create table h2 using hudi
+tblproperties (type = 'cow', primaryKey = 'id')
+partitioned by (dt)
+as
+select 1 as id, 'a1' as name, 10 as price, 1000 as dt;
+```
+
+**Example CTAS command to load data from another table.**
+
+```sql
+# create managed parquet table 
+create table parquet_mngd using parquet location 'file:///tmp/parquet_dataset/*.parquet';
+
+# CTAS by loading data into hudi table
+create table hudi_tbl using hudi location 'file:/tmp/hudi/hudi_tbl/' tblproperties ( 
+  type = 'cow', 
+  primaryKey = 'id', 
+  preCombineField = 'ts' 
+ ) 
+partitioned by (datestr) as select * from parquet_mngd;
+```
+
+### Set hoodie config options
+You can also set the config with table options when creating table which will work for
+the table scope only and override the config set by the SET command.
+```sql
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  ${hoodie.config.key1} = '${hoodie.config.value2}',
+  ${hoodie.config.key2} = '${hoodie.config.value2}',
+  ....
+);
+
+e.g.
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  hoodie.cleaner.fileversions.retained = '20',
+  hoodie.keep.max.commits = '20'
+);
+```
+
+## Spark Alter Table
+### Syntax
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+```
+
+:::note
+`ALTER TABLE ... RENAME TO ...` is not supported when using AWS Glue Data Catalog as hive metastore as Glue itself does 
+not support table renames.
+:::
+
+### Examples
+```sql
+alter table h0 rename to h0_1;
+
+alter table h0_1 add columns(ext0 string);
+
+alter table h0_1 change column id id bigint;
+```
+
+Schema evolution can be achieved via `ALTER TABLE` commands. Below shows some basic examples.
+
+:::note
+For more detailed examples, please prefer to [schema evolution](/docs/schema_evolution)
+:::
+
+**Syntax**
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+
+-- Alter table properties
+ALTER TABLE tableIdentifier SET TBLPROPERTIES (key = 'value')
+```
+
+**Examples**
+```sql
+--rename to:
+ALTER TABLE hudi_cow_nonpcf_tbl RENAME TO hudi_cow_nonpcf_tbl2;
+
+--add column:
+ALTER TABLE hudi_cow_nonpcf_tbl2 add columns(remark string);
+
+--change column:
+ALTER TABLE hudi_cow_nonpcf_tbl2 change column uuid uuid bigint;
+
+--set properties;
+alter table hudi_cow_nonpcf_tbl2 set tblproperties (hoodie.keep.max.commits = '10');
+```
+
+### Alter hoodie config options
+You can also alter the write config for a table by the **ALTER SERDEPROPERTIES**
+
+Example:
+```sql
+ alter table h3 set serdeproperties (hoodie.keep.max.commits = '10') 
+```
+
+### Use set command
+You can use the **set** command to set any custom hudi's config, which will work for the
+whole spark session scope.
+```sql
+set hoodie.insert.shuffle.parallelism = 100;
+set hoodie.upsert.shuffle.parallelism = 100;
+set hoodie.delete.shuffle.parallelism = 100;
+```
+
+### Show and Drop partitions
+
+**Syntax**
+
+```sql
+-- Show Partitions
+SHOW PARTITIONS tableIdentifier
+
+-- Drop Partition
+ALTER TABLE tableIdentifier DROP PARTITION ( partition_col_name = partition_col_val [ , ... ] )
+```
+
+**Examples**
+```sql
+--show partition:
+show partitions hudi_cow_pt_tbl;
+
+--drop partition：
+alter table hudi_cow_pt_tbl drop partition (dt='2021-12-09', hh='10');
+```
+
+:::note
+Currently,  the result of `show partitions` is based on the filesystem table path. It's not precise when delete the whole partition data or drop certain partition directly.
+:::
+
+## Flink
+
+### Create Catalog
+
+The catalog helps to manage the SQL tables, the table can be shared among CLI sessions if the catalog persists the table DDLs.
+For `hms` mode, the catalog also supplements the hive syncing options.
+
+HMS mode catalog SQL demo:

Review Comment:
   nit: "HMS mode catalog SQL demo" -> Example
   to keep it consistent.



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
+
+### Table Type
+Here is an example of creating a COW table.
+
+```sql
+-- create a non-primary key (or key less) table
+create table if not exists hudi_table2(
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow'
+);
+```
+
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 
+the table and hudi will treat them as key less table. Users can still perform Merge Into, updates and deletes based on any random data column. 
+
+### Primary Key
+Here is an example of creating COW table with a primary key 'id'. For mutable datasets, it is recommended to set appropriate primary key. 
+
+```sql
+-- create a managed cow table
+create table if not exists hudi_table0 (
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+);
+```
+
+### PreCombineField
+Here is an example of creating an MOR external table. The **preCombineField** option
+is used to specify the preCombine field for merge. Generally 'event time' or some other similar column will be used for 
+ordering purpose. Hudi will be able to handle out of order data using the precombine field value.
+
+```sql
+-- create an external mor table
+create table if not exists hudi_table1 (
+  id int, 
+  name string, 
+  price double,
+  ts bigint
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id,name',
+  preCombineField = 'ts' 
+);
+```
+
+### Partitioned Table
+Here is an example of creating a COW partitioned key less table.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+ ) 
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table
+Here is an example of creating a COW un-partitioned table.
+
+```sql
+-- create a cow table, with primaryKey 'uuid' and unpartitioned. 
+create table hudi_cow_nonpcf_tbl (
+  uuid int,
+  name string,
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'uuid'
+);
+```
+
+Here is an example of creating a MOR un-partitioned table.
+
+```sql
+-- create a mor non-partitioned table with preCombineField provided
+create table hudi_mor_tbl (
+    id int,
+    name string,
+    price double,
+    ts bigint
+) using hudi
+tblproperties (
+    type = 'mor',
+    primaryKey = 'id',
+    preCombineField = 'ts'
+);
+```
+
+### Create Table for an External Hudi Table
+You can create an External table using the `location` statement. If an external location is not specified it is considered a managed table. 
+You can read more about external vs managed tables [here](https://sparkbyexamples.com/apache-hive/difference-between-hive-internal-tables-and-external-tables/).
+An external table is useful if you need to read/write to/from a pre-existing hudi table.
+
+```sql
+ create table h_p1 using hudi
+ location '/path/to/hudi';
+```
+
+:::tip
+You don't need to specify schema and any properties except the partitioned columns if existed. Hudi can automatically recognize the schema and configurations.
+:::
+
+### Create Table AS SELECT
+
+Hudi supports CTAS(Create table as select) on spark sql. <br/>
+**Note:** For better performance to load data to hudi table, CTAS uses **bulk insert** as the write operation.
+
+**Example CTAS command to create a non-partitioned COW key less table.**
+
+```sql 
+create table h3 using hudi
+tblproperties (type = 'cow')
+as
+select 1 as id, 'a1' as name, 10 as price;
+```
+
+**Example CTAS command to create a partitioned, primary keyed COW table.**
+
+```sql
+create table h2 using hudi
+tblproperties (type = 'cow', primaryKey = 'id')
+partitioned by (dt)
+as
+select 1 as id, 'a1' as name, 10 as price, 1000 as dt;
+```
+
+**Example CTAS command to load data from another table.**
+
+```sql
+# create managed parquet table 
+create table parquet_mngd using parquet location 'file:///tmp/parquet_dataset/*.parquet';
+
+# CTAS by loading data into hudi table
+create table hudi_tbl using hudi location 'file:/tmp/hudi/hudi_tbl/' tblproperties ( 
+  type = 'cow', 
+  primaryKey = 'id', 
+  preCombineField = 'ts' 
+ ) 
+partitioned by (datestr) as select * from parquet_mngd;
+```
+
+### Set hoodie config options
+You can also set the config with table options when creating table which will work for
+the table scope only and override the config set by the SET command.
+```sql
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  ${hoodie.config.key1} = '${hoodie.config.value2}',
+  ${hoodie.config.key2} = '${hoodie.config.value2}',
+  ....
+);
+
+e.g.
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  hoodie.cleaner.fileversions.retained = '20',
+  hoodie.keep.max.commits = '20'
+);
+```
+
+## Spark Alter Table
+### Syntax
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+```
+
+:::note
+`ALTER TABLE ... RENAME TO ...` is not supported when using AWS Glue Data Catalog as hive metastore as Glue itself does 
+not support table renames.
+:::
+
+### Examples
+```sql
+alter table h0 rename to h0_1;
+
+alter table h0_1 add columns(ext0 string);
+
+alter table h0_1 change column id id bigint;
+```
+
+Schema evolution can be achieved via `ALTER TABLE` commands. Below shows some basic examples.
+
+:::note
+For more detailed examples, please prefer to [schema evolution](/docs/schema_evolution)
+:::
+
+**Syntax**
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+
+-- Alter table properties
+ALTER TABLE tableIdentifier SET TBLPROPERTIES (key = 'value')
+```
+
+**Examples**
+```sql
+--rename to:
+ALTER TABLE hudi_cow_nonpcf_tbl RENAME TO hudi_cow_nonpcf_tbl2;
+
+--add column:
+ALTER TABLE hudi_cow_nonpcf_tbl2 add columns(remark string);
+
+--change column:
+ALTER TABLE hudi_cow_nonpcf_tbl2 change column uuid uuid bigint;
+
+--set properties;
+alter table hudi_cow_nonpcf_tbl2 set tblproperties (hoodie.keep.max.commits = '10');
+```
+
+### Alter hoodie config options
+You can also alter the write config for a table by the **ALTER SERDEPROPERTIES**
+
+Example:
+```sql
+ alter table h3 set serdeproperties (hoodie.keep.max.commits = '10') 
+```
+
+### Use set command
+You can use the **set** command to set any custom hudi's config, which will work for the
+whole spark session scope.
+```sql
+set hoodie.insert.shuffle.parallelism = 100;
+set hoodie.upsert.shuffle.parallelism = 100;
+set hoodie.delete.shuffle.parallelism = 100;
+```
+
+### Show and Drop partitions
+
+**Syntax**
+
+```sql
+-- Show Partitions
+SHOW PARTITIONS tableIdentifier
+
+-- Drop Partition
+ALTER TABLE tableIdentifier DROP PARTITION ( partition_col_name = partition_col_val [ , ... ] )
+```
+
+**Examples**
+```sql
+--show partition:
+show partitions hudi_cow_pt_tbl;
+
+--drop partition：
+alter table hudi_cow_pt_tbl drop partition (dt='2021-12-09', hh='10');
+```
+
+:::note
+Currently,  the result of `show partitions` is based on the filesystem table path. It's not precise when delete the whole partition data or drop certain partition directly.
+:::
+
+## Flink
+
+### Create Catalog
+
+The catalog helps to manage the SQL tables, the table can be shared among CLI sessions if the catalog persists the table DDLs.
+For `hms` mode, the catalog also supplements the hive syncing options.

Review Comment:
   This statement is not clear. What is meant by `supplements the hive syncing options`?



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
+
+### Table Type
+Here is an example of creating a COW table.
+
+```sql
+-- create a non-primary key (or key less) table
+create table if not exists hudi_table2(
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow'
+);
+```
+
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 
+the table and hudi will treat them as key less table. Users can still perform Merge Into, updates and deletes based on any random data column. 
+
+### Primary Key
+Here is an example of creating COW table with a primary key 'id'. For mutable datasets, it is recommended to set appropriate primary key. 
+
+```sql
+-- create a managed cow table
+create table if not exists hudi_table0 (
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+);
+```
+
+### PreCombineField
+Here is an example of creating an MOR external table. The **preCombineField** option
+is used to specify the preCombine field for merge. Generally 'event time' or some other similar column will be used for 
+ordering purpose. Hudi will be able to handle out of order data using the precombine field value.
+
+```sql
+-- create an external mor table
+create table if not exists hudi_table1 (
+  id int, 
+  name string, 
+  price double,
+  ts bigint
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id,name',
+  preCombineField = 'ts' 
+);
+```
+
+### Partitioned Table
+Here is an example of creating a COW partitioned key less table.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+ ) 
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table
+Here is an example of creating a COW un-partitioned table.
+
+```sql
+-- create a cow table, with primaryKey 'uuid' and unpartitioned. 
+create table hudi_cow_nonpcf_tbl (
+  uuid int,
+  name string,
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'uuid'
+);
+```
+
+Here is an example of creating a MOR un-partitioned table.
+
+```sql
+-- create a mor non-partitioned table with preCombineField provided
+create table hudi_mor_tbl (
+    id int,
+    name string,
+    price double,
+    ts bigint
+) using hudi
+tblproperties (
+    type = 'mor',
+    primaryKey = 'id',
+    preCombineField = 'ts'
+);
+```
+
+### Create Table for an External Hudi Table
+You can create an External table using the `location` statement. If an external location is not specified it is considered a managed table. 
+You can read more about external vs managed tables [here](https://sparkbyexamples.com/apache-hive/difference-between-hive-internal-tables-and-external-tables/).
+An external table is useful if you need to read/write to/from a pre-existing hudi table.
+
+```sql
+ create table h_p1 using hudi
+ location '/path/to/hudi';
+```
+
+:::tip
+You don't need to specify schema and any properties except the partitioned columns if existed. Hudi can automatically recognize the schema and configurations.
+:::
+
+### Create Table AS SELECT
+
+Hudi supports CTAS(Create table as select) on spark sql. <br/>
+**Note:** For better performance to load data to hudi table, CTAS uses **bulk insert** as the write operation.
+
+**Example CTAS command to create a non-partitioned COW key less table.**
+
+```sql 
+create table h3 using hudi
+tblproperties (type = 'cow')
+as
+select 1 as id, 'a1' as name, 10 as price;
+```
+
+**Example CTAS command to create a partitioned, primary keyed COW table.**
+
+```sql
+create table h2 using hudi
+tblproperties (type = 'cow', primaryKey = 'id')
+partitioned by (dt)
+as
+select 1 as id, 'a1' as name, 10 as price, 1000 as dt;
+```
+
+**Example CTAS command to load data from another table.**
+
+```sql
+# create managed parquet table 
+create table parquet_mngd using parquet location 'file:///tmp/parquet_dataset/*.parquet';
+
+# CTAS by loading data into hudi table
+create table hudi_tbl using hudi location 'file:/tmp/hudi/hudi_tbl/' tblproperties ( 
+  type = 'cow', 
+  primaryKey = 'id', 
+  preCombineField = 'ts' 
+ ) 
+partitioned by (datestr) as select * from parquet_mngd;
+```
+
+### Set hoodie config options
+You can also set the config with table options when creating table which will work for
+the table scope only and override the config set by the SET command.
+```sql
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  ${hoodie.config.key1} = '${hoodie.config.value2}',
+  ${hoodie.config.key2} = '${hoodie.config.value2}',
+  ....
+);
+
+e.g.
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  hoodie.cleaner.fileversions.retained = '20',
+  hoodie.keep.max.commits = '20'
+);
+```
+
+## Spark Alter Table
+### Syntax
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+```
+
+:::note
+`ALTER TABLE ... RENAME TO ...` is not supported when using AWS Glue Data Catalog as hive metastore as Glue itself does 
+not support table renames.
+:::
+
+### Examples
+```sql
+alter table h0 rename to h0_1;
+
+alter table h0_1 add columns(ext0 string);
+
+alter table h0_1 change column id id bigint;
+```
+
+Schema evolution can be achieved via `ALTER TABLE` commands. Below shows some basic examples.
+
+:::note
+For more detailed examples, please prefer to [schema evolution](/docs/schema_evolution)
+:::
+
+**Syntax**
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+
+-- Alter table properties
+ALTER TABLE tableIdentifier SET TBLPROPERTIES (key = 'value')
+```
+
+**Examples**
+```sql
+--rename to:
+ALTER TABLE hudi_cow_nonpcf_tbl RENAME TO hudi_cow_nonpcf_tbl2;
+
+--add column:
+ALTER TABLE hudi_cow_nonpcf_tbl2 add columns(remark string);
+
+--change column:
+ALTER TABLE hudi_cow_nonpcf_tbl2 change column uuid uuid bigint;
+
+--set properties;
+alter table hudi_cow_nonpcf_tbl2 set tblproperties (hoodie.keep.max.commits = '10');
+```
+
+### Alter hoodie config options
+You can also alter the write config for a table by the **ALTER SERDEPROPERTIES**
+
+Example:
+```sql
+ alter table h3 set serdeproperties (hoodie.keep.max.commits = '10') 
+```
+
+### Use set command
+You can use the **set** command to set any custom hudi's config, which will work for the
+whole spark session scope.
+```sql
+set hoodie.insert.shuffle.parallelism = 100;
+set hoodie.upsert.shuffle.parallelism = 100;
+set hoodie.delete.shuffle.parallelism = 100;
+```
+
+### Show and Drop partitions
+
+**Syntax**
+
+```sql
+-- Show Partitions
+SHOW PARTITIONS tableIdentifier
+
+-- Drop Partition
+ALTER TABLE tableIdentifier DROP PARTITION ( partition_col_name = partition_col_val [ , ... ] )
+```
+
+**Examples**
+```sql
+--show partition:
+show partitions hudi_cow_pt_tbl;
+
+--drop partition：
+alter table hudi_cow_pt_tbl drop partition (dt='2021-12-09', hh='10');
+```
+
+:::note
+Currently,  the result of `show partitions` is based on the filesystem table path. It's not precise when delete the whole partition data or drop certain partition directly.
+:::
+
+## Flink
+
+### Create Catalog
+
+The catalog helps to manage the SQL tables, the table can be shared among CLI sessions if the catalog persists the table DDLs.

Review Comment:
   We can probably make it more clear. Is there some config to control persisting the DDLs? Under which modes is DDL persisting supported?



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
+
+### Table Type
+Here is an example of creating a COW table.
+
+```sql
+-- create a non-primary key (or key less) table
+create table if not exists hudi_table2(
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow'
+);
+```
+
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 
+the table and hudi will treat them as key less table. Users can still perform Merge Into, updates and deletes based on any random data column. 
+
+### Primary Key
+Here is an example of creating COW table with a primary key 'id'. For mutable datasets, it is recommended to set appropriate primary key. 
+
+```sql
+-- create a managed cow table
+create table if not exists hudi_table0 (
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+);
+```
+
+### PreCombineField
+Here is an example of creating an MOR external table. The **preCombineField** option
+is used to specify the preCombine field for merge. Generally 'event time' or some other similar column will be used for 
+ordering purpose. Hudi will be able to handle out of order data using the precombine field value.
+
+```sql
+-- create an external mor table
+create table if not exists hudi_table1 (
+  id int, 
+  name string, 
+  price double,
+  ts bigint
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id,name',
+  preCombineField = 'ts' 
+);
+```
+
+### Partitioned Table
+Here is an example of creating a COW partitioned key less table.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+ ) 
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table
+Here is an example of creating a COW un-partitioned table.
+
+```sql
+-- create a cow table, with primaryKey 'uuid' and unpartitioned. 
+create table hudi_cow_nonpcf_tbl (
+  uuid int,
+  name string,
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'uuid'
+);
+```
+
+Here is an example of creating a MOR un-partitioned table.
+
+```sql
+-- create a mor non-partitioned table with preCombineField provided
+create table hudi_mor_tbl (
+    id int,
+    name string,
+    price double,
+    ts bigint
+) using hudi
+tblproperties (
+    type = 'mor',
+    primaryKey = 'id',
+    preCombineField = 'ts'
+);
+```
+
+### Create Table for an External Hudi Table
+You can create an External table using the `location` statement. If an external location is not specified it is considered a managed table. 
+You can read more about external vs managed tables [here](https://sparkbyexamples.com/apache-hive/difference-between-hive-internal-tables-and-external-tables/).
+An external table is useful if you need to read/write to/from a pre-existing hudi table.
+
+```sql
+ create table h_p1 using hudi
+ location '/path/to/hudi';
+```
+
+:::tip
+You don't need to specify schema and any properties except the partitioned columns if existed. Hudi can automatically recognize the schema and configurations.
+:::
+
+### Create Table AS SELECT
+
+Hudi supports CTAS(Create table as select) on spark sql. <br/>
+**Note:** For better performance to load data to hudi table, CTAS uses **bulk insert** as the write operation.
+
+**Example CTAS command to create a non-partitioned COW key less table.**
+
+```sql 
+create table h3 using hudi
+tblproperties (type = 'cow')
+as
+select 1 as id, 'a1' as name, 10 as price;
+```
+
+**Example CTAS command to create a partitioned, primary keyed COW table.**
+
+```sql
+create table h2 using hudi
+tblproperties (type = 'cow', primaryKey = 'id')
+partitioned by (dt)
+as
+select 1 as id, 'a1' as name, 10 as price, 1000 as dt;
+```
+
+**Example CTAS command to load data from another table.**
+
+```sql
+# create managed parquet table 
+create table parquet_mngd using parquet location 'file:///tmp/parquet_dataset/*.parquet';
+
+# CTAS by loading data into hudi table
+create table hudi_tbl using hudi location 'file:/tmp/hudi/hudi_tbl/' tblproperties ( 
+  type = 'cow', 
+  primaryKey = 'id', 
+  preCombineField = 'ts' 
+ ) 
+partitioned by (datestr) as select * from parquet_mngd;
+```
+
+### Set hoodie config options
+You can also set the config with table options when creating table which will work for
+the table scope only and override the config set by the SET command.
+```sql
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  ${hoodie.config.key1} = '${hoodie.config.value2}',
+  ${hoodie.config.key2} = '${hoodie.config.value2}',
+  ....
+);
+
+e.g.
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  hoodie.cleaner.fileversions.retained = '20',
+  hoodie.keep.max.commits = '20'
+);
+```
+
+## Spark Alter Table
+### Syntax
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+```
+
+:::note
+`ALTER TABLE ... RENAME TO ...` is not supported when using AWS Glue Data Catalog as hive metastore as Glue itself does 
+not support table renames.
+:::
+
+### Examples
+```sql
+alter table h0 rename to h0_1;
+
+alter table h0_1 add columns(ext0 string);
+
+alter table h0_1 change column id id bigint;
+```
+
+Schema evolution can be achieved via `ALTER TABLE` commands. Below shows some basic examples.
+
+:::note
+For more detailed examples, please prefer to [schema evolution](/docs/schema_evolution)
+:::
+
+**Syntax**
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+
+-- Alter table properties
+ALTER TABLE tableIdentifier SET TBLPROPERTIES (key = 'value')
+```
+
+**Examples**
+```sql
+--rename to:
+ALTER TABLE hudi_cow_nonpcf_tbl RENAME TO hudi_cow_nonpcf_tbl2;
+
+--add column:
+ALTER TABLE hudi_cow_nonpcf_tbl2 add columns(remark string);
+
+--change column:
+ALTER TABLE hudi_cow_nonpcf_tbl2 change column uuid uuid bigint;
+
+--set properties;
+alter table hudi_cow_nonpcf_tbl2 set tblproperties (hoodie.keep.max.commits = '10');
+```
+
+### Alter hoodie config options
+You can also alter the write config for a table by the **ALTER SERDEPROPERTIES**
+
+Example:
+```sql
+ alter table h3 set serdeproperties (hoodie.keep.max.commits = '10') 
+```
+
+### Use set command
+You can use the **set** command to set any custom hudi's config, which will work for the
+whole spark session scope.
+```sql
+set hoodie.insert.shuffle.parallelism = 100;
+set hoodie.upsert.shuffle.parallelism = 100;
+set hoodie.delete.shuffle.parallelism = 100;
+```
+
+### Show and Drop partitions
+
+**Syntax**
+
+```sql
+-- Show Partitions
+SHOW PARTITIONS tableIdentifier
+
+-- Drop Partition
+ALTER TABLE tableIdentifier DROP PARTITION ( partition_col_name = partition_col_val [ , ... ] )
+```
+
+**Examples**
+```sql
+--show partition:
+show partitions hudi_cow_pt_tbl;
+
+--drop partition：
+alter table hudi_cow_pt_tbl drop partition (dt='2021-12-09', hh='10');
+```
+
+:::note
+Currently,  the result of `show partitions` is based on the filesystem table path. It's not precise when delete the whole partition data or drop certain partition directly.

Review Comment:
   nit: The second statement `It's not precise when delete the whole partition data or drop certain partition directly.` is not clear. Do you mean to say `It's not consistent when deleting the whole partition data or dropping a partition directly.`



##########
website/docs/sql_ddl.md:
##########
@@ -0,0 +1,409 @@
+---
+title: SQL DDL
+summary: "In this page, we introduce how to create tables with Hudi."
+toc: true
+last_modified_at: 
+---
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+The following are SparkSQL DDL actions available:
+
+## Spark Create Table
+:::note
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+:::
+
+### Options
+
+Users can set table options while creating a hudi table.
+
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
+
+To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
+
+### Table Type
+Here is an example of creating a COW table.
+
+```sql
+-- create a non-primary key (or key less) table
+create table if not exists hudi_table2(
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow'
+);
+```
+
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 
+the table and hudi will treat them as key less table. Users can still perform Merge Into, updates and deletes based on any random data column. 
+
+### Primary Key
+Here is an example of creating COW table with a primary key 'id'. For mutable datasets, it is recommended to set appropriate primary key. 
+
+```sql
+-- create a managed cow table
+create table if not exists hudi_table0 (
+  id int, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+);
+```
+
+### PreCombineField
+Here is an example of creating an MOR external table. The **preCombineField** option
+is used to specify the preCombine field for merge. Generally 'event time' or some other similar column will be used for 
+ordering purpose. Hudi will be able to handle out of order data using the precombine field value.
+
+```sql
+-- create an external mor table
+create table if not exists hudi_table1 (
+  id int, 
+  name string, 
+  price double,
+  ts bigint
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id,name',
+  preCombineField = 'ts' 
+);
+```
+
+### Partitioned Table
+Here is an example of creating a COW partitioned key less table.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'cow',
+  primaryKey = 'id'
+ ) 
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table
+Here is an example of creating a COW un-partitioned table.
+
+```sql
+-- create a cow table, with primaryKey 'uuid' and unpartitioned. 
+create table hudi_cow_nonpcf_tbl (
+  uuid int,
+  name string,
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'uuid'
+);
+```
+
+Here is an example of creating a MOR un-partitioned table.
+
+```sql
+-- create a mor non-partitioned table with preCombineField provided
+create table hudi_mor_tbl (
+    id int,
+    name string,
+    price double,
+    ts bigint
+) using hudi
+tblproperties (
+    type = 'mor',
+    primaryKey = 'id',
+    preCombineField = 'ts'
+);
+```
+
+### Create Table for an External Hudi Table
+You can create an External table using the `location` statement. If an external location is not specified it is considered a managed table. 
+You can read more about external vs managed tables [here](https://sparkbyexamples.com/apache-hive/difference-between-hive-internal-tables-and-external-tables/).
+An external table is useful if you need to read/write to/from a pre-existing hudi table.
+
+```sql
+ create table h_p1 using hudi
+ location '/path/to/hudi';
+```
+
+:::tip
+You don't need to specify schema and any properties except the partitioned columns if existed. Hudi can automatically recognize the schema and configurations.
+:::
+
+### Create Table AS SELECT
+
+Hudi supports CTAS(Create table as select) on spark sql. <br/>
+**Note:** For better performance to load data to hudi table, CTAS uses **bulk insert** as the write operation.
+
+**Example CTAS command to create a non-partitioned COW key less table.**
+
+```sql 
+create table h3 using hudi
+tblproperties (type = 'cow')
+as
+select 1 as id, 'a1' as name, 10 as price;
+```
+
+**Example CTAS command to create a partitioned, primary keyed COW table.**
+
+```sql
+create table h2 using hudi
+tblproperties (type = 'cow', primaryKey = 'id')
+partitioned by (dt)
+as
+select 1 as id, 'a1' as name, 10 as price, 1000 as dt;
+```
+
+**Example CTAS command to load data from another table.**
+
+```sql
+# create managed parquet table 
+create table parquet_mngd using parquet location 'file:///tmp/parquet_dataset/*.parquet';
+
+# CTAS by loading data into hudi table
+create table hudi_tbl using hudi location 'file:/tmp/hudi/hudi_tbl/' tblproperties ( 
+  type = 'cow', 
+  primaryKey = 'id', 
+  preCombineField = 'ts' 
+ ) 
+partitioned by (datestr) as select * from parquet_mngd;
+```
+
+### Set hoodie config options
+You can also set the config with table options when creating table which will work for
+the table scope only and override the config set by the SET command.
+```sql
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  ${hoodie.config.key1} = '${hoodie.config.value2}',
+  ${hoodie.config.key2} = '${hoodie.config.value2}',
+  ....
+);
+
+e.g.
+create table if not exists h3(
+  id bigint, 
+  name string, 
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'id',
+  type = 'mor',
+  hoodie.cleaner.fileversions.retained = '20',
+  hoodie.keep.max.commits = '20'
+);
+```
+
+## Spark Alter Table
+### Syntax
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+```
+
+:::note
+`ALTER TABLE ... RENAME TO ...` is not supported when using AWS Glue Data Catalog as hive metastore as Glue itself does 
+not support table renames.
+:::
+
+### Examples
+```sql
+alter table h0 rename to h0_1;
+
+alter table h0_1 add columns(ext0 string);
+
+alter table h0_1 change column id id bigint;
+```
+
+Schema evolution can be achieved via `ALTER TABLE` commands. Below shows some basic examples.
+
+:::note
+For more detailed examples, please prefer to [schema evolution](/docs/schema_evolution)
+:::
+
+**Syntax**
+```sql
+-- Alter table name
+ALTER TABLE oldTableName RENAME TO newTableName
+
+-- Alter table add columns
+ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+
+-- Alter table column type
+ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+
+-- Alter table properties
+ALTER TABLE tableIdentifier SET TBLPROPERTIES (key = 'value')
+```
+
+**Examples**
+```sql
+--rename to:
+ALTER TABLE hudi_cow_nonpcf_tbl RENAME TO hudi_cow_nonpcf_tbl2;
+
+--add column:
+ALTER TABLE hudi_cow_nonpcf_tbl2 add columns(remark string);
+
+--change column:
+ALTER TABLE hudi_cow_nonpcf_tbl2 change column uuid uuid bigint;
+
+--set properties;
+alter table hudi_cow_nonpcf_tbl2 set tblproperties (hoodie.keep.max.commits = '10');
+```
+
+### Alter hoodie config options
+You can also alter the write config for a table by the **ALTER SERDEPROPERTIES**
+
+Example:
+```sql
+ alter table h3 set serdeproperties (hoodie.keep.max.commits = '10') 
+```
+
+### Use set command
+You can use the **set** command to set any custom hudi's config, which will work for the
+whole spark session scope.
+```sql
+set hoodie.insert.shuffle.parallelism = 100;
+set hoodie.upsert.shuffle.parallelism = 100;
+set hoodie.delete.shuffle.parallelism = 100;
+```
+
+### Show and Drop partitions
+
+**Syntax**
+
+```sql
+-- Show Partitions
+SHOW PARTITIONS tableIdentifier
+
+-- Drop Partition
+ALTER TABLE tableIdentifier DROP PARTITION ( partition_col_name = partition_col_val [ , ... ] )
+```
+
+**Examples**
+```sql
+--show partition:
+show partitions hudi_cow_pt_tbl;
+
+--drop partition：
+alter table hudi_cow_pt_tbl drop partition (dt='2021-12-09', hh='10');
+```
+
+:::note
+Currently,  the result of `show partitions` is based on the filesystem table path. It's not precise when delete the whole partition data or drop certain partition directly.
+:::
+
+## Flink
+
+### Create Catalog
+
+The catalog helps to manage the SQL tables, the table can be shared among CLI sessions if the catalog persists the table DDLs.
+For `hms` mode, the catalog also supplements the hive syncing options.
+
+HMS mode catalog SQL demo:
+```sql
+CREATE CATALOG hoodie_catalog
+  WITH (
+    'type'='hudi',
+    'catalog.path' = '${catalog default root path}',
+    'hive.conf.dir' = '${directory where hive-site.xml is located}',
+    'mode'='hms' -- supports 'dfs' mode that uses the DFS backend for table DDLs persistence
+  );
+```
+
+#### Options
+|  Option Name  | Required | Default | Remarks |
+|  -----------  | -------  | ------- | ------- |
+| `catalog.path` | true | -- | Default root path for the catalog, the path is used to infer the table path automatically, the default table path: `${catalog.path}/${db_name}/${table_name}` |
+| `default-database` | false | default | default database name |
+| `hive.conf.dir` | false | -- | The directory where hive-site.xml is located, only valid in `hms` mode |
+| `mode` | false | dfs | Supports `hms` mode that uses HMS to persist the table options |
+| `table.external` | false | false | Whether to create the external table, only valid in `hms` mode |
+
+### Create Table
+
+The following is a Flink example to create a table. [Read the Flink Quick Start](/docs/flink-quick-start-guide) guide for more examples.
+
+```sql 
+CREATE TABLE hudi_table2(
+  id int, 
+  name string, 
+  price double
+)
+WITH (
+'connector' = 'hudi',
+'path' = 's3://bucket-name/hudi/',
+'table.type' = 'MERGE_ON_READ' -- this creates a MERGE_ON_READ table, by default is COPY_ON_WRITE

Review Comment:
   nit: remove "by"



##########
website/docs/sql_dml.md:
##########
@@ -69,56 +74,110 @@ create table if not exists hudi_table1 (
   price double,
   ts bigint
 ) using hudi
-options (
+tblproperties (
   type = 'mor',
   primaryKey = 'id,name',
   preCombineField = 'ts' 
 );
 ```
 
 ### Partitioned Table
-Here is an example of creating a COW partitioned table.
+Here is an example of creating a COW partitioned key less table.

Review Comment:
   It is not a keyless table.



##########
website/docs/sql_dml.md:
##########
@@ -7,42 +7,46 @@ last_modified_at:
 import Tabs from '@theme/Tabs';
 import TabItem from '@theme/TabItem';
 
-The following are SparkSQL table management actions available:
+The following are SparkSQL DDL actions available:

Review Comment:
   nit: The following DDL actions are available via SparkSQL



##########
website/docs/sql_dml.md:
##########
@@ -7,42 +7,46 @@ last_modified_at:
 import Tabs from '@theme/Tabs';
 import TabItem from '@theme/TabItem';
 
-The following are SparkSQL table management actions available:
+The following are SparkSQL DDL actions available:
 
 ## Spark Create Table
 :::note
-Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
+Only SparkSQL needs an explicit Create Table command. No Create Table command is required in Spark when using Scala or 
+Python. The first batch of a [Write](/docs/writing_data) to a table will create the table if it does not exist.
 :::
 
 ### Options
 
 Users can set table options while creating a hudi table.
 
-| Parameter Name | Description | (Optional/Required) : Default Value |
-|------------|--------|--------|
-| primaryKey | The primary key names of the table, multiple fields separated by commas. | (Optional) : `id`|
-| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.| (Optional) : `cow` |
-| preCombineField | The Pre-Combine field of the table. | (Optional) : `ts`|
+| Parameter Name | Description                                                                                                                                                                                                        | (Optional/Required) : Default Value |
+|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|
+| primaryKey | The primary key names of the table, multiple fields separated by commas. When set, hudi will ensure uniqueness during updates and deletes. When this config is skipped, hudi treats the table as a key less table. | (Optional) : `id`|
+| type       | The type of table to create ([read more](/docs/table_types)). <br></br> `cow` = COPY-ON-WRITE, `mor` = MERGE-ON-READ.                                                                                              | (Optional) : `cow` |
+| preCombineField | The Pre-Combine field of the table. This field will be used in resolving the final version of the record when two versions are combined with merges or updates.                                                    | (Optional) : `ts`|
 
 To set any custom hudi config(like index type, max parquet size, etc), see the  "Set hudi config section" .
 
 ### Table Type
 Here is an example of creating a COW table.
 
 ```sql
--- create a non-primary key table
+-- create a non-primary key (or key less) table
 create table if not exists hudi_table2(
   id int, 
   name string, 
   price double
 ) using hudi
-options (
+tblproperties (
   type = 'cow'
 );
 ```
 
+There could be datasets where primary key may not be feasible. For such use-cases, user don't need to elect any primary key for 

Review Comment:
   nit: user -> users



##########
website/docs/sql_dml.md:
##########
@@ -69,56 +74,110 @@ create table if not exists hudi_table1 (
   price double,
   ts bigint
 ) using hudi
-options (
+tblproperties (

Review Comment:
   Please correct the comment, it is not an external table



##########
website/docs/sql_dml.md:
##########
@@ -69,56 +74,110 @@ create table if not exists hudi_table1 (
   price double,
   ts bigint
 ) using hudi
-options (
+tblproperties (
   type = 'mor',
   primaryKey = 'id,name',
   preCombineField = 'ts' 
 );
 ```
 
 ### Partitioned Table
-Here is an example of creating a COW partitioned table.
+Here is an example of creating a COW partitioned key less table.
 ```sql
 create table if not exists hudi_table_p0 (
 id bigint,
 name string,
 dt string,
 hh string  
 ) using hudi
-options (
+tblproperties (
   type = 'cow',
   primaryKey = 'id'
  ) 
-partitioned by (dt, hh);
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table
+Here is an example of creating a COW un-partitioned table.
+
+```sql
+-- create a cow table, with primaryKey 'uuid' and unpartitioned. 

Review Comment:
   -- create a non-partitioned cow table with primaryKey 'uuid'



##########
website/docs/sql_dml.md:
##########
@@ -69,56 +74,110 @@ create table if not exists hudi_table1 (
   price double,
   ts bigint
 ) using hudi
-options (
+tblproperties (
   type = 'mor',
   primaryKey = 'id,name',
   preCombineField = 'ts' 
 );
 ```
 
 ### Partitioned Table
-Here is an example of creating a COW partitioned table.
+Here is an example of creating a COW partitioned key less table.
 ```sql
 create table if not exists hudi_table_p0 (
 id bigint,
 name string,
 dt string,
 hh string  
 ) using hudi
-options (
+tblproperties (
   type = 'cow',
   primaryKey = 'id'
  ) 
-partitioned by (dt, hh);
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table
+Here is an example of creating a COW un-partitioned table.
+
+```sql
+-- create a cow table, with primaryKey 'uuid' and unpartitioned. 
+create table hudi_cow_nonpcf_tbl (
+  uuid int,
+  name string,
+  price double
+) using hudi
+tblproperties (
+  primaryKey = 'uuid'
+);
+```
+
+Here is an example of creating a MOR un-partitioned table.

Review Comment:
   ditto



##########
website/docs/quick-start-guide.md:
##########
@@ -216,9 +237,33 @@ can generate sample inserts and updates based on the the sample trip schema [her
 
 ## Create Table
 
+Before we go further, few terminologies to familiarize: 
+
+- **Table types**
+
+  Hudi supports two different table types, namely Copy-On-Write (COW) and Merge-On-Read (MOR). Users can choose either 
+of these table types depending on their workload and SLA requirements. You can read more about different 
+  table types [here](/docs/next/table_types/).
+
+- **Partitioned & Non-Partitioned tables**
+
+  Users can create a partitioned table or a non-partitioned table with Apache Hudi. Partitioning can help with 
+  reducing query run times. For quick start purpose, we will go with partitioned table. 
+
+- **Primary key and Hudi table**
+
+  Optionally users can choose to create a Primary keyed table. When primary key is set for a given table,
+  Hudi ensures uniqueness during updates and deletes. Each record is uniquely identified by the primary key configuration.
+  If primary key is not set, Hudi treats it as key less table and every record ingested is treated as a new record even
+  if contents match. Such keyless tables are supported from Hudi 0.14.0.
+
+:::note
+For the purpose of quick start guide, we will go with one table type (COW) which is partitioned.

Review Comment:
   nit: we will go with only COW table type which is partitioned.



##########
website/docs/quick-start-guide.md:
##########
@@ -1718,5 +1209,5 @@ Also, we used Spark here to show case the capabilities of Hudi. However, Hudi ca
 Hudi tables can be queried from query engines like Hive, Spark, Presto and much more. We have put together a 
 [demo video](https://www.youtube.com/watch?v=VhNgUsxdrD0) that show cases all of this on a docker based setup with all 
 dependent systems running locally. We recommend you replicate the same setup and run the demo yourself, by following 
-steps [here](/docs/docker_demo) to get a taste for it. Also, if you are looking for ways to migrate your existing data 
-to Hudi, refer to [migration guide](/docs/migration_guide).
+steps [here](/docs/next/docker_demo) to get a taste for it. Also, if you are looking for ways to migrate your existing data 

Review Comment:
   nit: taste of it.



##########
website/docs/sql_dml.md:
##########
@@ -69,56 +74,110 @@ create table if not exists hudi_table1 (
   price double,
   ts bigint
 ) using hudi
-options (
+tblproperties (
   type = 'mor',
   primaryKey = 'id,name',
   preCombineField = 'ts' 
 );
 ```
 
 ### Partitioned Table
-Here is an example of creating a COW partitioned table.
+Here is an example of creating a COW partitioned key less table.
 ```sql
 create table if not exists hudi_table_p0 (
 id bigint,
 name string,
 dt string,
 hh string  
 ) using hudi
-options (
+tblproperties (
   type = 'cow',
   primaryKey = 'id'
  ) 
-partitioned by (dt, hh);
+partitioned by dt;
+```
+
+Here is an example of creating a MOR partitioned table with preCombine field.
+```sql
+create table if not exists hudi_table_p0 (
+id bigint,
+name string,
+dt string,
+hh string  
+) using hudi
+tblproperties (
+  type = 'mor',
+  primaryKey = 'id',
+  preCombineField = 'ts'
+ ) 
+partitioned by dt;
+```
+
+### Un-Partitioned Table
+Here is an example of creating a COW un-partitioned table.

Review Comment:
   ditto



##########
website/docs/quick-start-guide.md:
##########
@@ -1538,174 +998,205 @@ values={[
 <TabItem value="scala">
 
 ```scala
-// spark-shell
-spark.
-  read.format("hudi").
-  load(basePath).
-  select("uuid","partitionpath").
-  sort("partitionpath","uuid").
-  show(100, false)
+spark.read.
+  format("hudi").
+  option("as.of.instant", "20210728141108100").
+  load(basePath)
 
-val inserts = convertToStringList(dataGen.generateInserts(10))
-val df = spark.
-  read.json(spark.sparkContext.parallelize(inserts, 2)).
-  filter("partitionpath = 'americas/united_states/san_francisco'")
-df.write.format("hudi").
-  options(getQuickstartWriteConfigs).
-  option(OPERATION.key(),"insert_overwrite").
-  option(PRECOMBINE_FIELD.key(), "ts").
-  option(RECORDKEY_FIELD.key(), "uuid").
-  option(PARTITIONPATH_FIELD.key(), "partitionpath").
-  option(TBL_NAME.key(), tableName).
-  mode(Append).
-  save(basePath)
+spark.read.
+  format("hudi").
+  option("as.of.instant", "2021-07-28 14:11:08.200").
+  load(basePath)
+
+// It is equal to "as.of.instant = 2021-07-28 00:00:00"
+spark.read.
+  format("hudi").
+  option("as.of.instant", "2021-07-28").
+  load(basePath)
 
-// Should have different keys now for San Francisco alone, from query before.
-spark.
-  read.format("hudi").
-  load(basePath).
-  select("uuid","partitionpath").
-  sort("partitionpath","uuid").
-  show(100, false)
 ```
-</TabItem>
 
+</TabItem>
 <TabItem value="python">
 
 ```python
-# pyspark
-spark.read.format("hudi"). \
-    load(basePath). \
-    select(["uuid", "partitionpath"]). \
-    sort(["partitionpath", "uuid"]). \
-    show(n=100, truncate=False)
-    
-inserts = sc._jvm.org.apache.hudi.QuickstartUtils.convertToStringList(dataGen.generateInserts(10)) 
-df = spark.read.json(spark.sparkContext.parallelize(inserts, 2)). \
-    filter("partitionpath = 'americas/united_states/san_francisco'")
-hudi_insert_overwrite_options = {
-    'hoodie.table.name': tableName,
-    'hoodie.datasource.write.recordkey.field': 'uuid',
-    'hoodie.datasource.write.partitionpath.field': 'partitionpath',
-    'hoodie.datasource.write.table.name': tableName,
-    'hoodie.datasource.write.operation': 'insert_overwrite',
-    'hoodie.datasource.write.precombine.field': 'ts',
-    'hoodie.upsert.shuffle.parallelism': 2,
-    'hoodie.insert.shuffle.parallelism': 2
-}
-df.write.format("hudi").options(**hudi_insert_overwrite_options).mode("append").save(basePath)
-spark.read.format("hudi"). \
-    load(basePath). \
-    select(["uuid", "partitionpath"]). \
-    sort(["partitionpath", "uuid"]). \
-    show(n=100, truncate=False)
+#pyspark
+spark.read. \
+  format("hudi"). \
+  option("as.of.instant", "20210728141108"). \
+  load(basePath)
+
+spark.read. \
+  format("hudi"). \
+  option("as.of.instant", "2021-07-28 14:11:08.000"). \
+  load(basePath)
+
+# It is equal to "as.of.instant = 2021-07-28 00:00:00"
+spark.read. \
+  format("hudi"). \
+  option("as.of.instant", "2021-07-28"). \
+  load(basePath)
 ```
-</TabItem>
 
+</TabItem>
 <TabItem value="sparksql">
 
-`insert overwrite` a partitioned table use the `INSERT_OVERWRITE` type of write operation, while a non-partitioned table to `INSERT_OVERWRITE_TABLE`.
+:::note
+Requires Spark 3.2+
+:::
 
 ```sql
--- insert overwrite non-partitioned table
-insert overwrite hudi_mor_tbl select 99, 'a99', 20.0, 900;
-insert overwrite hudi_cow_nonpcf_tbl select 99, 'a99', 20.0;
 
--- insert overwrite partitioned table with dynamic partition
-insert overwrite table hudi_cow_pt_tbl select 10, 'a10', 1100, '2021-12-09', '10';
-
--- insert overwrite partitioned table with static partition
-insert overwrite hudi_cow_pt_tbl partition(dt = '2021-12-09', hh='12') select 13, 'a13', 1100;
+-- time travel based on commit time, for eg: `20220307091628793`
+select * from hudi_cow_pt_tbl timestamp as of '20220307091628793' where id = 1;
+-- time travel based on different timestamp formats
+select * from hudi_cow_pt_tbl timestamp as of '2022-03-07 09:16:28.100' where id = 1;
+select * from hudi_cow_pt_tbl timestamp as of '2022-03-08' where id = 1;
 ```
+
 </TabItem>
 
 </Tabs
 >
 
-## More Spark SQL Commands
 
-### Alter Table
+### Incremental query
 
-Schema evolution can be achieved via `ALTER TABLE` commands. Below shows some basic examples.
+Hudi also provides capability to obtain a stream of records that changed since given commit timestamp. 
+This can be achieved using Hudi's incremental querying and providing a begin time from which changes need to be streamed. 
+We do not need to specify endTime, if we want all changes after the given commit (as is the common case). 
 
-:::note
-For more detailed examples, please prefer to [schema evolution](schema_evolution)
-:::
+<Tabs
+groupId="programming-language"
+defaultValue="scala"
+values={[
+{ label: 'Scala', value: 'scala', },
+{ label: 'Python', value: 'python', },
+{ label: 'Spark SQL', value: 'sparksql', }
+]}
+>
 
-**Syntax**
-```sql
--- Alter table name
-ALTER TABLE oldTableName RENAME TO newTableName
+<TabItem value="scala">
+
+```scala
+// spark-shell
+// reload data
+spark.
+  read.
+  format("hudi").
+  load(basePath).
+  createOrReplaceTempView("hudi_trips_snapshot")
 
--- Alter table add columns
-ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
+val commits = spark.sql("select distinct(_hoodie_commit_time) as commitTime from  hudi_trips_snapshot order by commitTime").map(k => k.getString(0)).take(50)
+val beginTime = commits(commits.length - 2) // commit time we are interested in
 
--- Alter table column type
-ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
+// incrementally query data
+val tripsIncrementalDF = spark.read.format("hudi").
+  option(QUERY_TYPE.key(), QUERY_TYPE_INCREMENTAL_OPT_VAL).
+  option(BEGIN_INSTANTTIME.key(), beginTime).
+  load(basePath)
+tripsIncrementalDF.createOrReplaceTempView("hudi_trips_incremental")
 
--- Alter table properties
-ALTER TABLE tableIdentifier SET TBLPROPERTIES (key = 'value')
+spark.sql("select `_hoodie_commit_time`, fare, begin_lon, begin_lat, ts from  hudi_trips_incremental where fare > 20.0").show()
 ```
-**Examples**
-```sql
---rename to:
-ALTER TABLE hudi_cow_nonpcf_tbl RENAME TO hudi_cow_nonpcf_tbl2;
 
---add column:
-ALTER TABLE hudi_cow_nonpcf_tbl2 add columns(remark string);
+</TabItem>
+<TabItem value="python">
+
+```python
+# pyspark
+# reload data
+spark. \
+  read. \
+  format("hudi"). \
+  load(basePath). \
+  createOrReplaceTempView("hudi_trips_snapshot")
+
+commits = list(map(lambda row: row[0], spark.sql("select distinct(_hoodie_commit_time) as commitTime from  hudi_trips_snapshot order by commitTime").limit(50).collect()))
+beginTime = commits[len(commits) - 2] # commit time we are interested in
+
+# incrementally query data
+incremental_read_options = {
+  'hoodie.datasource.query.type': 'incremental',
+  'hoodie.datasource.read.begin.instanttime': beginTime,
+}
 
---change column:
-ALTER TABLE hudi_cow_nonpcf_tbl2 change column uuid uuid bigint;
+tripsIncrementalDF = spark.read.format("hudi"). \
+  options(**incremental_read_options). \
+  load(basePath)
+tripsIncrementalDF.createOrReplaceTempView("hudi_trips_incremental")
 
---set properties;
-alter table hudi_cow_nonpcf_tbl2 set tblproperties (hoodie.keep.max.commits = '10');
+spark.sql("select `_hoodie_commit_time`, fare, begin_lon, begin_lat, ts from  hudi_trips_incremental where fare > 20.0").show()
 ```
 
-### Partition SQL Command
+</TabItem>
 
-**Syntax**
+<TabItem value="sparksql">
 
 ```sql
--- Add Partition
-ALTER TABLE tableIdentifier ADD PARTITION ( partition_col_name = partition_col_val [ , ... ] )
+-- syntax
+hudi_table_changes(table or path, queryType, beginTime [, endTime]);  
+-- table or path: table identifier, example: db.tableName, tableName, 
+--                or path for of your table, example: path/to/hudiTable  
+--                in this case table does not need to exist in the metastore,
+-- queryType: incremental query mode, example: latest_state, cdc  
+--            (for cdc query, first enable cdc for your table by setting cdc.enabled=true),
+-- beginTime: instantTime to begin query from, example: earliest, 202305150000, 
+-- endTime: optional instantTime to end query at, example: 202305160000, 
 
--- Drop Partition
-ALTER TABLE tableIdentifier DROP PARTITION ( partition_col_name = partition_col_val [ , ... ] )
+-- incrementally query data by table name
+-- start from earliest available commit, end at latest available commit.  
+select * from hudi_table_changes('db.table', 'latest_state', 'earliest');
 
--- Show Partitions
-SHOW PARTITIONS tableIdentifier
-```
-**Examples**
-```sql
---show partition:
-show partitions hudi_cow_pt_tbl;
+-- start from earliest, end at 202305160000.  
+select * from hudi_table_changes('table', 'latest_state', 'earliest', '202305160000');  
+
+-- start from 202305150000, end at 202305160000.
+select * from hudi_table_changes('table', 'latest_state', '202305150000', '202305160000');
+
+-- incrementally query data by path
+-- start from earliest available commit, end at latest available commit.
+select * from hudi_table_changes('path/to/table', 'cdc', 'earliest');
+
+-- start from earliest, end at 202305160000.
+select * from hudi_table_changes('path/to/table', 'cdc', 'earliest', '202305160000');
+
+-- start from 202305150000, end at 202305160000.
+select * from hudi_table_changes('path/to/table', 'cdc', '202305150000', '202305160000');
 
---drop partition：
-alter table hudi_cow_pt_tbl drop partition (dt='2021-12-09', hh='10');
 ```
-:::note
-Currently,  the result of `show partitions` is based on the filesystem table path. It's not precise when delete the whole partition data or drop certain partition directly.
 
+</TabItem>
+
+</Tabs
+>
+
+:::info
+This will give all changes that happened after the beginTime commit with the filter of fare > 20.0. The unique thing about this
+feature is that it now lets you author streaming pipelines on batch data.
 :::
 
-### Procedures
+## Streaming writers
+### Hudi Streamer
+Hudi provides a ingestion tool to assist with ingesting data into hudi from various difference sources in a streaming manner. 
+This has lot of niceties like auto checkpointing, schema enforcement via schema provider, transformation support and so on.

Review Comment:
   Learnt a new word 'nicety' :) 



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [hudi] bhasudha commented on pull request #9712: [HUDI-6851] Fixing Spark quick start guide

Posted by "bhasudha (via GitHub)" <gi...@apache.org>.

bhasudha commented on PR #9712:
URL: https://github.com/apache/hudi/pull/9712#issuecomment-1720195813

   Left some comments on the other PR - https://github.com/apache/hudi/pull/9622. and synced with @nsivabalan  to ensure those are addressed here. 


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org