text: {
    "plan" : [ {
      "@type" : "ksqlPlanV1",
      "statementText" : "CREATE STREAM INPUT_STREAM (K STRING KEY, SF BIGINT) WITH (KAFKA_TOPIC='stream_topic', VALUE_FORMAT='JSON');",
      "ddlCommand" : {
        "@type" : "createStreamV1",
        "sourceName" : "INPUT_STREAM",
        "schema" : "`K` STRING KEY, `SF` BIGINT",
        "topicName" : "stream_topic",
        "formats" : {
          "keyFormat" : {
            "format" : "KAFKA"
          },
          "valueFormat" : {
            "format" : "JSON"
          }
        }
      }
    }, {
      "@type" : "ksqlPlanV1",
      "statementText" : "CREATE TABLE INPUT_TABLE (ID BIGINT PRIMARY KEY, TF INTEGER) WITH (KAFKA_TOPIC='table_topic', VALUE_FORMAT='JSON');",
      "ddlCommand" : {
        "@type" : "createTableV1",
        "sourceName" : "INPUT_TABLE",
        "schema" : "`ID` BIGINT KEY, `TF` INTEGER",
        "topicName" : "table_topic",
        "formats" : {
          "keyFormat" : {
            "format" : "KAFKA"
          },
          "valueFormat" : {
            "format" : "JSON"
          }
        }
      }
    }, {
      "@type" : "ksqlPlanV1",
      "statementText" : "CREATE STREAM OUTPUT AS SELECT\n  *,\n  S.ROWTIME S_ROWTIME,\n  T.ROWTIME T_ROWTIME\nFROM INPUT_STREAM S\nINNER JOIN INPUT_TABLE T ON ((S.SF = T.ID))\nEMIT CHANGES",
      "ddlCommand" : {
        "@type" : "createStreamV1",
        "sourceName" : "OUTPUT",
        "schema" : "`S_SF` BIGINT KEY, `S_K` STRING, `T_ID` BIGINT, `T_TF` INTEGER, `S_ROWTIME` BIGINT, `T_ROWTIME` BIGINT",
        "topicName" : "OUTPUT",
        "formats" : {
          "keyFormat" : {
            "format" : "KAFKA"
          },
          "valueFormat" : {
            "format" : "JSON"
          }
        }
      },
      "queryPlan" : {
        "sources" : [ "INPUT_STREAM", "INPUT_TABLE" ],
        "sink" : "OUTPUT",
        "physicalPlan" : {
          "@type" : "streamSinkV1",
          "properties" : {
            "queryContext" : "OUTPUT"
          },
          "source" : {
            "@type" : "streamSelectV1",
            "properties" : {
              "queryContext" : "Project"
            },
            "source" : {
              "@type" : "streamTableJoinV1",
              "properties" : {
                "queryContext" : "Join"
              },
              "joinType" : "INNER",
              "internalFormats" : {
                "keyFormat" : {
                  "format" : "KAFKA"
                },
                "valueFormat" : {
                  "format" : "JSON"
                }
              },
              "leftSource" : {
                "@type" : "streamSelectV1",
                "properties" : {
                  "queryContext" : "PrependAliasLeft"
                },
                "source" : {
                  "@type" : "streamSelectKeyV2",
                  "properties" : {
                    "queryContext" : "LeftSourceKeyed"
                  },
                  "source" : {
                    "@type" : "streamSourceV1",
                    "properties" : {
                      "queryContext" : "KafkaTopic_Left/Source"
                    },
                    "topicName" : "stream_topic",
                    "formats" : {
                      "keyFormat" : {
                        "format" : "KAFKA"
                      },
                      "valueFormat" : {
                        "format" : "JSON"
                      }
                    },
                    "sourceSchema" : "`K` STRING KEY, `SF` BIGINT"
                  },
                  "keyExpression" : "SF"
                },
                "keyColumnNames" : [ "S_SF" ],
                "selectExpressions" : [ "SF AS S_SF", "ROWTIME AS S_ROWTIME", "K AS S_K" ]
              },
              "rightSource" : {
                "@type" : "tableSelectV1",
                "properties" : {
                  "queryContext" : "PrependAliasRight"
                },
                "source" : {
                  "@type" : "tableSourceV1",
                  "properties" : {
                    "queryContext" : "KafkaTopic_Right/Source"
                  },
                  "topicName" : "table_topic",
                  "formats" : {
                    "keyFormat" : {
                      "format" : "KAFKA"
                    },
                    "valueFormat" : {
                      "format" : "JSON"
                    }
                  },
                  "sourceSchema" : "`ID` BIGINT KEY, `TF` INTEGER"
                },
                "keyColumnNames" : [ "T_ID" ],
                "selectExpressions" : [ "TF AS T_TF", "ROWTIME AS T_ROWTIME", "ID AS T_ID" ]
              },
              "keyColName" : "S_SF"
            },
            "keyColumnNames" : [ "S_SF" ],
            "selectExpressions" : [ "S_K AS S_K", "T_ID AS T_ID", "T_TF AS T_TF", "S_ROWTIME AS S_ROWTIME", "T_ROWTIME AS T_ROWTIME" ]
          },
          "formats" : {
            "keyFormat" : {
              "format" : "KAFKA"
            },
            "valueFormat" : {
              "format" : "JSON"
            }
          },
          "topicName" : "OUTPUT"
        },
        "queryId" : "CSAS_OUTPUT_0"
      }
    } ],
    "configs" : {
      "ksql.extension.dir" : "ext",
      "ksql.streams.cache.max.bytes.buffering" : "0",
      "ksql.security.extension.class" : null,
      "ksql.transient.prefix" : "transient_",
      "ksql.persistence.wrap.single.values" : "true",
      "ksql.authorization.cache.expiry.time.secs" : "30",
      "ksql.schema.registry.url" : "",
      "ksql.streams.default.deserialization.exception.handler" : "io.confluent.ksql.errors.LogMetricAndContinueExceptionHandler",
      "ksql.output.topic.name.prefix" : "",
      "ksql.streams.auto.offset.reset" : "earliest",
      "ksql.query.pull.enable.standby.reads" : "false",
      "ksql.connect.url" : "http://localhost:8083",
      "ksql.service.id" : "some.ksql.service.id",
      "ksql.internal.topic.min.insync.replicas" : "1",
      "ksql.streams.shutdown.timeout.ms" : "300000",
      "ksql.internal.topic.replicas" : "1",
      "ksql.insert.into.values.enabled" : "true",
      "ksql.query.pull.max.allowed.offset.lag" : "9223372036854775807",
      "ksql.query.pull.max.qps" : "2147483647",
      "ksql.streams.default.production.exception.handler" : "io.confluent.ksql.errors.ProductionExceptionHandlerUtil$LogAndFailProductionExceptionHandler",
      "ksql.access.validator.enable" : "auto",
      "ksql.streams.bootstrap.servers" : "localhost:0",
      "ksql.streams.commit.interval.ms" : "2000",
      "ksql.metric.reporters" : "",
      "ksql.query.pull.metrics.enabled" : "false",
      "ksql.streams.auto.commit.interval.ms" : "0",
      "ksql.metrics.extension" : null,
      "ksql.streams.topology.optimization" : "all",
      "ksql.hidden.topics" : "_confluent.*,__confluent.*,_schemas,__consumer_offsets,__transaction_state,connect-configs,connect-offsets,connect-status,connect-statuses",
      "ksql.streams.num.stream.threads" : "4",
      "ksql.timestamp.throw.on.invalid" : "false",
      "ksql.authorization.cache.max.entries" : "10000",
      "ksql.metrics.tags.custom" : "",
      "ksql.pull.queries.enable" : "true",
      "ksql.udfs.enabled" : "true",
      "ksql.udf.enable.security.manager" : "true",
      "ksql.connect.worker.config" : "",
      "ksql.sink.window.change.log.additional.retention" : "1000000",
      "ksql.readonly.topics" : "_confluent.*,__confluent.*,_schemas,__consumer_offsets,__transaction_state,connect-configs,connect-offsets,connect-status,connect-statuses",
      "ksql.udf.collect.metrics" : "false",
      "ksql.persistent.prefix" : "query_",
      "ksql.query.persistent.active.limit" : "2147483647"
    }
  }