Join is a binary logical operator, i.e. works with two logical operators. Join has a join type and an optional expression condition for the join.
Join is created when…FIXME
|
Note
|
CROSS JOIN is just an INNER JOIN with no join condition. |
Join has output schema attributes…FIXME
Join takes the following when created:
Logical plan of the left side
Logical plan of the right side
Join condition (if available) as a Catalyst expression
InsertIntoTable is a unary logical operator that represents the following:
INSERT INTO and INSERT OVERWRITE TABLE SQL statements
DataFrameWriter is requested to insert the rows of a DataFrame into a table
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
// make sure that the tables are available in a catalog sql("CREATE TABLE IF NOT EXISTS t1(id long)") sql("CREATE TABLE IF NOT EXISTS t2(id long)") val q = sql("INSERT INTO TABLE t2 SELECT * from t1 LIMIT 100") val plan = q.queryExecution.logical scala> println(plan.numberedTreeString) 00 'InsertIntoTable 'UnresolvedRelation `t2`, false, false 01 +- 'GlobalLimit 100 02 +- 'LocalLimit 100 03 +- 'Project [*] 04 +- 'UnresolvedRelation `t1` // Dataset API's version of "INSERT OVERWRITE TABLE" in SQL spark.range(10).write.mode("overwrite").insertInto("t2") |
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
INSERT INTO partitioned_table spark.range(10) .withColumn("p1", 'id % 2) .write .mode("overwrite") .partitionBy("p1") .saveAsTable("partitioned_table") val insertIntoQ = sql("INSERT INTO TABLE partitioned_table PARTITION (p1 = 4) VALUES 41, 42") scala> println(insertIntoQ.queryExecution.logical.numberedTreeString) 00 'InsertIntoTable 'UnresolvedRelation `partitioned_table`, Map(p1 -> Some(4)), false, false 01 +- 'UnresolvedInlineTable [col1], [List(41), List(42)] |
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
INSERT OVERWRITE TABLE partitioned_table spark.range(10) .withColumn("p1", 'id % 2) .write .mode("overwrite") .partitionBy("p1") .saveAsTable("partitioned_table") val insertOverwriteQ = sql("INSERT OVERWRITE TABLE partitioned_table PARTITION (p1 = 4) VALUES 40") scala> println(insertOverwriteQ.queryExecution.logical.numberedTreeString) 00 'InsertIntoTable 'UnresolvedRelation `partitioned_table`, Map(p1 -> Some(4)), true, false 01 +- 'UnresolvedInlineTable [col1], [List(40)] |
InsertIntoTable is created with partition keys that correspond to the partitionSpec part of the following SQL statements:
INSERT INTO TABLE (with the overwrite and ifPartitionNotExists flags off)
INSERT OVERWRITE TABLE (with the overwrite and ifPartitionNotExists flags off)
InsertIntoTable has no partition keys when created as follows:
insertInto operator from the Catalyst DSL
DataFrameWriter.insertInto operator
InsertIntoTable can never be resolved (i.e. InsertIntoTable should not be part of a logical plan after analysis and is supposed to be converted to logical commands at analysis phase).
| Logical Command | Description |
|---|---|
|
When HiveAnalysis resolution rule transforms |
|
|
When DataSourceAnalysis posthoc logical resolution resolves an |
|
|
When DataSourceAnalysis posthoc logical resolution transforms |
|
Caution
|
FIXME What’s the difference between HiveAnalysis that converts InsertIntoTable(r: HiveTableRelation…) to InsertIntoHiveTable and RelationConversions that converts InsertIntoTable(r: HiveTableRelation,…) to InsertIntoTable (with LogicalRelation)?
|
|
Note
|
Inserting into views or RDD-based tables is not allowed (and fails at analysis). |
InsertIntoTable (with UnresolvedRelation leaf logical operator) is created when:
INSERT INTO or INSERT OVERWRITE TABLE SQL statements are executed (as a single insert or a multi-insert query)
DataFrameWriter is requested to insert a DataFrame into a table
RelationConversions logical evaluation rule is executed (and transforms InsertIntoTable operators)
CreateHiveTableAsSelectCommand logical command is executed
InsertIntoTable has an empty output schema.
|
Tip
|
Use
|
InsertIntoTable takes the following when created:
Logical plan for the table to insert into
Partition keys (with optional partition values for dynamic partition insert)
Logical plan representing the data to be written
overwrite flag that indicates whether to overwrite an existing table or partitions (true) or not (false)
Inserting into a view is not allowed, i.e. a query plan with an InsertIntoTable operator with a UnresolvedRelation leaf operator that is resolved to a View unary operator fails at analysis (when ResolveRelations logical resolution is executed).
|
1 2 3 4 5 |
Inserting into a view is not allowed. View: [name]. |
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 |
// Create a view val viewName = "demo_view" sql(s"DROP VIEW IF EXISTS $viewName") assert(spark.catalog.tableExists(viewName) == false) sql(s"CREATE VIEW $viewName COMMENT 'demo view' AS SELECT 1,2,3") assert(spark.catalog.tableExists(viewName)) // The following should fail with an AnalysisException scala> spark.range(0).write.insertInto(viewName) org.apache.spark.sql.AnalysisException: Inserting into a view is not allowed. View: `default`.`demo_view`.; at org.apache.spark.sql.catalyst.analysis.package$AnalysisErrorAt.failAnalysis(package.scala:42) at org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveRelations$$anonfun$apply$8.applyOrElse(Analyzer.scala:644) at org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveRelations$$anonfun$apply$8.applyOrElse(Analyzer.scala:640) at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformUp$1.apply(TreeNode.scala:289) at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformUp$1.apply(TreeNode.scala:289) at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:70) at org.apache.spark.sql.catalyst.trees.TreeNode.transformUp(TreeNode.scala:288) at org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveRelations$.apply(Analyzer.scala:640) at org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveRelations$.apply(Analyzer.scala:586) at org.apache.spark.sql.catalyst.rules.RuleExecutor$$anonfun$execute$1$$anonfun$apply$1.apply(RuleExecutor.scala:87) at org.apache.spark.sql.catalyst.rules.RuleExecutor$$anonfun$execute$1$$anonfun$apply$1.apply(RuleExecutor.scala:84) at scala.collection.LinearSeqOptimized$class.foldLeft(LinearSeqOptimized.scala:124) at scala.collection.immutable.List.foldLeft(List.scala:84) at org.apache.spark.sql.catalyst.rules.RuleExecutor$$anonfun$execute$1.apply(RuleExecutor.scala:84) at org.apache.spark.sql.catalyst.rules.RuleExecutor$$anonfun$execute$1.apply(RuleExecutor.scala:76) at scala.collection.immutable.List.foreach(List.scala:381) at org.apache.spark.sql.catalyst.rules.RuleExecutor.execute(RuleExecutor.scala:76) at org.apache.spark.sql.catalyst.analysis.Analyzer.org$apache$spark$sql$catalyst$analysis$Analyzer$$executeSameContext(Analyzer.scala:124) at org.apache.spark.sql.catalyst.analysis.Analyzer.execute(Analyzer.scala:118) at org.apache.spark.sql.catalyst.analysis.Analyzer.executeAndCheck(Analyzer.scala:103) at org.apache.spark.sql.execution.QueryExecution.analyzed$lzycompute(QueryExecution.scala:57) at org.apache.spark.sql.execution.QueryExecution.analyzed(QueryExecution.scala:55) at org.apache.spark.sql.execution.QueryExecution.assertAnalyzed(QueryExecution.scala:47) at org.apache.spark.sql.execution.QueryExecution.withCachedData$lzycompute(QueryExecution.scala:61) at org.apache.spark.sql.execution.QueryExecution.withCachedData(QueryExecution.scala:60) at org.apache.spark.sql.execution.QueryExecution.optimizedPlan$lzycompute(QueryExecution.scala:66) at org.apache.spark.sql.execution.QueryExecution.optimizedPlan(QueryExecution.scala:66) at org.apache.spark.sql.execution.QueryExecution.sparkPlan$lzycompute(QueryExecution.scala:72) at org.apache.spark.sql.execution.QueryExecution.sparkPlan(QueryExecution.scala:68) at org.apache.spark.sql.execution.QueryExecution.executedPlan$lzycompute(QueryExecution.scala:77) at org.apache.spark.sql.execution.QueryExecution.executedPlan(QueryExecution.scala:77) at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:75) at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:654) at org.apache.spark.sql.DataFrameWriter.insertInto(DataFrameWriter.scala:322) at org.apache.spark.sql.DataFrameWriter.insertInto(DataFrameWriter.scala:308) ... 49 elided |
Inserting into an RDD-based table is not allowed, i.e. a query plan with an InsertIntoTable operator with one of the following logical operators (as the logical plan representing the table) fails at analysis (when PreWriteCheck extended logical check is executed):
Logical operator is not a leaf node
Range leaf operator
OneRowRelation leaf operator
LocalRelation leaf operator
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 |
// Create a temporary view val data = spark.range(1) data.createOrReplaceTempView("demo") scala> spark.range(0).write.insertInto("demo") org.apache.spark.sql.AnalysisException: Inserting into an RDD-based table is not allowed.;; 'InsertIntoTable Range (0, 1, step=1, splits=Some(8)), false, false +- Range (0, 0, step=1, splits=Some(8)) at org.apache.spark.sql.execution.datasources.PreWriteCheck$.failAnalysis(rules.scala:442) at org.apache.spark.sql.execution.datasources.PreWriteCheck$$anonfun$apply$14.apply(rules.scala:473) at org.apache.spark.sql.execution.datasources.PreWriteCheck$$anonfun$apply$14.apply(rules.scala:445) at org.apache.spark.sql.catalyst.trees.TreeNode.foreach(TreeNode.scala:117) at org.apache.spark.sql.execution.datasources.PreWriteCheck$.apply(rules.scala:445) at org.apache.spark.sql.execution.datasources.PreWriteCheck$.apply(rules.scala:440) at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$2.apply(CheckAnalysis.scala:349) at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$2.apply(CheckAnalysis.scala:349) at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48) at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$class.checkAnalysis(CheckAnalysis.scala:349) at org.apache.spark.sql.catalyst.analysis.Analyzer.checkAnalysis(Analyzer.scala:92) at org.apache.spark.sql.catalyst.analysis.Analyzer.executeAndCheck(Analyzer.scala:105) at org.apache.spark.sql.execution.QueryExecution.analyzed$lzycompute(QueryExecution.scala:57) at org.apache.spark.sql.execution.QueryExecution.analyzed(QueryExecution.scala:55) at org.apache.spark.sql.execution.QueryExecution.assertAnalyzed(QueryExecution.scala:47) at org.apache.spark.sql.execution.QueryExecution.withCachedData$lzycompute(QueryExecution.scala:61) at org.apache.spark.sql.execution.QueryExecution.withCachedData(QueryExecution.scala:60) at org.apache.spark.sql.execution.QueryExecution.optimizedPlan$lzycompute(QueryExecution.scala:66) at org.apache.spark.sql.execution.QueryExecution.optimizedPlan(QueryExecution.scala:66) at org.apache.spark.sql.execution.QueryExecution.sparkPlan$lzycompute(QueryExecution.scala:72) at org.apache.spark.sql.execution.QueryExecution.sparkPlan(QueryExecution.scala:68) at org.apache.spark.sql.execution.QueryExecution.executedPlan$lzycompute(QueryExecution.scala:77) at org.apache.spark.sql.execution.QueryExecution.executedPlan(QueryExecution.scala:77) at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:75) at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:654) at org.apache.spark.sql.DataFrameWriter.insertInto(DataFrameWriter.scala:322) at org.apache.spark.sql.DataFrameWriter.insertInto(DataFrameWriter.scala:308) ... 49 elided |
InsertIntoHiveTable is…FIXME
spark技术分享run Method|
1 2 3 4 5 |
run(session: SparkSession): Seq[Row] |
|
Note
|
run is part of RunnableCommand Contract to execute (run) a logical command.
|
run…FIXME
processInsert Internal Method|
1 2 3 4 5 6 7 8 9 10 11 |
processInsert( sparkSession: SparkSession, externalCatalog: ExternalCatalog, hadoopConf: Configuration, tableDesc: TableDesc, tmpLocation: Path, child: SparkPlan): Unit |
processInsert…FIXME
|
Note
|
processInsert is used exclusively when InsertIntoHiveTable logical command is executed.
|
InsertIntoHiveDirCommand is…FIXME
run Method|
1 2 3 4 5 |
run(session: SparkSession): Seq[Row] |
|
Note
|
run is part of RunnableCommand Contract to execute (run) a logical command.
|
run…FIXME
InsertIntoHadoopFsRelationCommand is a DataWritingCommand that writes the data of the query out using the FileFormat.
InsertIntoHadoopFsRelationCommand is created when:
DataSource is requested to planForWritingFileFormat
DataSourceAnalysis post-hoc logical resolution rule is executed (and resolves a InsertIntoTable logical operator with a HadoopFsRelation)
run Method|
1 2 3 4 5 |
run(session: SparkSession): Seq[Row] |
|
Note
|
run is part of RunnableCommand Contract to execute (run) a logical command.
|
run uses the spark.sql.hive.manageFilesourcePartitions configuration property to…FIXME
|
Caution
|
FIXME When is the catalogTable defined? |
|
Caution
|
FIXME When is tracksPartitionsInCatalog of CatalogTable enabled?
|
run…FIXME
run uses SchemaUtils to make sure that there are no duplicates in the outputColumnNames.
InsertIntoHadoopFsRelationCommand takes the following when created:
Output Hadoop’s Path
|
Note
|
staticPartitions may hold zero or more partitions as follows:
With that, staticPartitions are simply the partitions of an InsertIntoTable logical operator. |
InsertIntoDataSourceDirCommand is a logical command that FIXME.
run Method|
1 2 3 4 5 |
run(sparkSession: SparkSession): Seq[Row] |
|
Note
|
run is part of RunnableCommand Contract to execute (run) a logical command.
|
run…FIXME
InsertIntoDataSourceCommand is a RunnableCommand that inserts or overwrites data in an InsertableRelation (per overwrite flag).
InsertIntoDataSourceCommand is created exclusively when DataSourceAnalysis logical resolution is executed (and resolves an InsertIntoTable unary logical operator with a LogicalRelation on an InsertableRelation).
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
sql("DROP TABLE IF EXISTS t2") sql("CREATE TABLE t2(id long)") val query = "SELECT * FROM RANGE(1)" // Using INSERT INTO SQL statement so we can access QueryExecution // DataFrameWriter.insertInto returns no value val q = sql("INSERT INTO TABLE t2 " + query) val logicalPlan = q.queryExecution.logical scala> println(logicalPlan.numberedTreeString) 00 'InsertIntoTable 'UnresolvedRelation `t2`, false, false 01 +- 'Project [*] 02 +- 'UnresolvedTableValuedFunction RANGE, [1] val analyzedPlan = q.queryExecution.analyzed scala> println(analyzedPlan.numberedTreeString) 00 InsertIntoHiveTable `default`.`t2`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, false, false, [id#6L] 01 +- Project [id#6L] 02 +- Range (0, 1, step=1, splits=None) |
InsertIntoDataSourceCommand returns the logical query plan when requested for the inner nodes (that should be shown as an inner nested tree of this node).
|
1 2 3 4 5 6 7 8 9 10 11 |
val query = "SELECT * FROM RANGE(1)" val sqlText = "INSERT INTO TABLE t2 " + query val plan = spark.sessionState.sqlParser.parsePlan(sqlText) scala> println(plan.numberedTreeString) 00 'InsertIntoTable 'UnresolvedRelation `t2`, false, false 01 +- 'Project [*] 02 +- 'UnresolvedTableValuedFunction RANGE, [1] |
run Method|
1 2 3 4 5 |
run(session: SparkSession): Seq[Row] |
|
Note
|
run is part of RunnableCommand Contract to execute (run) a logical command.
|
run takes the InsertableRelation (that is the relation of the LogicalRelation).
run then creates a DataFrame for the logical query plan and the input SparkSession.
run requests the DataFrame for the QueryExecution that in turn is requested for the RDD (of the structured query). run requests the LogicalRelation for the output schema.
With the RDD and the output schema, run creates another DataFrame that is the RDD[InternalRow] with the schema applied.
run requests the InsertableRelation to insert or overwrite data.
In the end, since the data in the InsertableRelation has changed, run requests the CacheManager to recacheByPlan with the LogicalRelation.
|
Note
|
run requests the SparkSession for SharedState that is in turn requested for the CacheManager.
|
InsertIntoDataSourceCommand takes the following when created:
LogicalRelation leaf logical operator
InMemoryRelation is a leaf logical operator that represents a cached child physical query plan.
InMemoryRelation is created when:
Dataset.persist operator is used (and in turn requests CacheManager to cache a structured query)
CatalogImpl is requested to cache a table or view in-memory or refreshTable
InsertIntoDataSourceCommand logical command is executed (and in turn requests CacheManager to recacheByPlan)
CatalogImpl is requested to refreshByPath (and in turn requests CacheManager to recacheByPath)
QueryExecution is requested for a cached logical query plan (and in turn requests CacheManager to replace logical query segments with cached query plans)
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 |
// Cache sample table range5 using pure SQL // That registers range5 to contain the output of range(5) function spark.sql("CACHE TABLE range5 AS SELECT * FROM range(5)") val q1 = spark.sql("SELECT * FROM range5") scala> q1.explain == Physical Plan == InMemoryTableScan [id#0L] +- InMemoryRelation [id#0L], true, 10000, StorageLevel(disk, memory, deserialized, 1 replicas), `range5` +- *Range (0, 5, step=1, splits=8) // you could also use optimizedPlan to see InMemoryRelation scala> println(q1.queryExecution.optimizedPlan.numberedTreeString) 00 InMemoryRelation [id#0L], true, 10000, StorageLevel(disk, memory, deserialized, 1 replicas), `range5` 01 +- *Range (0, 5, step=1, splits=8) // Use Dataset's cache val q2 = spark.range(10).groupBy('id % 5).count.cache scala> println(q2.queryExecution.optimizedPlan.numberedTreeString) 00 InMemoryRelation [(id % 5)#84L, count#83L], true, 10000, StorageLevel(disk, memory, deserialized, 1 replicas) 01 +- *HashAggregate(keys=[(id#77L % 5)#88L], functions=[count(1)], output=[(id % 5)#84L, count#83L]) 02 +- Exchange hashpartitioning((id#77L % 5)#88L, 200) 03 +- *HashAggregate(keys=[(id#77L % 5) AS (id#77L % 5)#88L], functions=[partial_count(1)], output=[(id#77L % 5)#88L, count#90L]) 04 +- *Range (0, 10, step=1, splits=8) |
InMemoryRelation is a MultiInstanceRelation so a new instance will be created to appear multiple times in a physical query plan.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 |
// Cache a Dataset val q = spark.range(10).cache // Make sure that q Dataset is cached val cache = spark.sharedState.cacheManager scala> cache.lookupCachedData(q.queryExecution.logical).isDefined res0: Boolean = true scala> q.explain == Physical Plan == InMemoryTableScan [id#122L] +- InMemoryRelation [id#122L], true, 10000, StorageLevel(disk, memory, deserialized, 1 replicas) +- *Range (0, 10, step=1, splits=8) val qCrossJoined = q.crossJoin(q) scala> println(qCrossJoined.queryExecution.optimizedPlan.numberedTreeString) 00 Join Cross 01 :- InMemoryRelation [id#122L], true, 10000, StorageLevel(disk, memory, deserialized, 1 replicas) 02 : +- *Range (0, 10, step=1, splits=8) 03 +- InMemoryRelation [id#170L], true, 10000, StorageLevel(disk, memory, deserialized, 1 replicas) 04 +- *Range (0, 10, step=1, splits=8) // Use sameResult for comparison // since the plans use different output attributes // and have to be canonicalized internally import org.apache.spark.sql.execution.columnar.InMemoryRelation val optimizedPlan = qCrossJoined.queryExecution.optimizedPlan scala> optimizedPlan.children(0).sameResult(optimizedPlan.children(1)) res1: Boolean = true |
|
Note
|
|
| Name | Description |
|---|---|
|
PartitionStatistics for the output schema Used exclusively when |
computeStats Method|
1 2 3 4 5 |
computeStats(): Statistics |
|
Note
|
computeStats is part of LeafNode Contract to compute statistics for cost-based optimizer.
|
computeStats…FIXME
InMemoryRelation takes the following when created:
Output schema attributes
Child physical query plan
Statistics of the child query plan
withOutput Method|
1 2 3 4 5 |
withOutput(newOutput: Seq[Attribute]): InMemoryRelation |
withOutput…FIXME
|
Note
|
withOutput is used exclusively when CacheManager is requested to replace logical query segments with cached query plans.
|
newInstance Method|
1 2 3 4 5 |
newInstance(): this.type |
|
Note
|
newInstance is part of MultiInstanceRelation Contract to…FIXME.
|
newInstance…FIXME
cachedColumnBuffers Method|
1 2 3 4 5 |
cachedColumnBuffers: RDD[CachedBatch] |
cachedColumnBuffers…FIXME
|
Note
|
cachedColumnBuffers is used when…FIXME
|
PartitionStatistics|
1 2 3 4 5 |
PartitionStatistics(tableSchema: Seq[Attribute]) |
|
Note
|
PartitionStatistics is a private[columnar] class.
|
PartitionStatistics…FIXME
|
Note
|
PartitionStatistics is used exclusively when InMemoryRelation is created (and initializes partitionStatistics).
|
