You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by GitBox <gi...@apache.org> on 2020/08/01 12:43:49 UTC
[GitHub] [incubator-pinot] mangrrua opened a new pull request #5787: [Feature] - Spark Pinot Connector
mangrrua opened a new pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787
## Description
Pinot is a great tool for OLAP queries in real-time. In many cases, users want to see aggregated results in realtime. In the real scenarios, all data do not come in real-time, or some data/reports should be calculated in batch(every hour, day etc).
[Apache Spark](https://spark.apache.org/) is a great tool for batch computing, data preparation and etc in ETL processes. And a lot of companies are using Apache Spark for their purposes.
Apache Spark and Apache Pinot are great with together for data preparation, aggregation, and query for many cases. But integrations between tools(eg spark-pinot) are so important! Pinot has a `spark-batch-ingestion` module, but it requires some efforts, and i know many developers like me do not want to do these efforts. These efforts are;
- User should analyze data with spark(eg), then it should save outputs to hdfs with parquet, orc or etc format.
- Trigger an `spark-batch-ingestion` job to convert analyzed results to offline segments. And wait for finish(also some trigger mechanisms and fail scenarios etc).
- `spark-batch-ingestion` reads input files analyzed above(each of these files represents one segment, and if user want to partition data in pinot, these input files must be partitioned), and creates segments, then writes it to the deep storage.
Unnecessary step is so painful! But not finished!
What happens we want to re-index data, or apply more aggregation in Pinot? For example, we want to re-index orders data by another dimension or etc. Or maybe another department in your company want to access your pinot data? Yes, all steps should be applied from zero.
What is the suggested solution? A connector that read/write from/to pinot directly.
If we can read/write data to pinot from spark directly, we can;
- Pinot(source) -> Spark(analyze) -> Pinot(sink)
- Pinot(source) -> Spark(analyze) -> Somewhere(hdfs, cassandra, postgres etc)
- Somewhere(hdfs, cassandra, postgres etc) -> Spark(analyze) -> Pinot(sink)
We can solve some problems with spark connector.
Connector supports only read for now. We are waiting new segment write API for write operation to prevent duplicate effort. This is the just initial version. In the future, `streaming endpoints` will applied for read, and write API will be added.
For this version, look the [pinot-spark-connector README](https://github.com/mangrrua/incubator-pinot/blob/spark-pinot-connector/pinot-connectors/pinot-spark-connector/README.md) and [read-model documentation](https://github.com/mangrrua/incubator-pinot/blob/spark-pinot-connector/pinot-connectors/pinot-spark-connector/documentation/read_model.md) for detailed info(I will move documentations https://docs.pinot.apache.org/developers/developers-and-contributors/update-document).
**Note:** Presto is the powerful engine for joins or other operations, but it does not the same with spark! Uses cases are different. Just focus spark ecosystem and etl processes.
Please share your comments and improvements.
## Upgrade Notes
Does this PR prevent a zero down-time upgrade? (Assume upgrade order: Controller, Broker, Server, Minion)
* [ ] Yes (Please label as **<code>backward-incompat</code>**, and complete the section below on Release Notes)
Does this PR fix a zero-downtime upgrade introduced earlier?
* [ ] Yes (Please label this as **<code>backward-incompat</code>**, and complete the section below on Release Notes)
Does this PR otherwise need attention when creating release notes? Things to consider:
- New configuration options
- Deprecation of configurations
- Signature changes to public methods/interfaces
- New plugins added or old plugins removed
* [ ] Yes (Please label this PR as **<code>release-notes</code>** and complete the section on Release Notes)
## Release Notes
If you have tagged this as either backward-incompat or release-notes,
you MUST add text here that you would like to see appear in release notes of the
next release.
If you have a series of commits adding or enabling a feature, then
add this section only in final commit that marks the feature completed.
Refer to earlier release notes to see examples of text
## Documentation
If you have introduced a new feature or configuration, please add it to the documentation as well.
See https://docs.pinot.apache.org/developers/developers-and-contributors/update-document
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r463990289
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/query/GeneratedPQLs.scala
##########
@@ -0,0 +1,33 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector.query
+
+import org.apache.pinot.connector.spark.connector.Constants.{PinotTableType, PinotTableTypes}
+
+private[pinot] case class GeneratedPQLs(
Review comment:
pql was changed with sql
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua edited a comment on pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua edited a comment on pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#issuecomment-672613922
> @fx19880617 @mangrrua anything pending on this?
Seems no @kishoreg
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] kishoreg commented on pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
kishoreg commented on pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#issuecomment-672287954
@fx19880617 @mangrrua anything pending on this?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] flykent1990 commented on pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
flykent1990 commented on pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#issuecomment-738851196
hi @mangrrua
"streaming endpoints" for spark connector currently this feature is available ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#issuecomment-672613922
> @fx19880617 @mangrrua anything pending on this?
Seems no
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua edited a comment on pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua edited a comment on pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#issuecomment-667951838
> This is great work! Looking forward to streaming API for read and writeAPI :)
>
> One question maybe related to this, shall we create a PinotRDD?
Thank you :)
Spark DatasourceV2 API supports very flexibility and features. We do not need to create an RDD. We are working on the dataframe and dataset api. After read data from pinot, all you have to the is play in spark
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r464326792
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/PinotServerDataFetcher.scala
##########
@@ -0,0 +1,134 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+import java.util.{List => JList, Map => JMap}
+
+import com.yammer.metrics.core.MetricsRegistry
+import org.apache.helix.model.InstanceConfig
+import org.apache.pinot.common.metrics.BrokerMetrics
+import org.apache.pinot.common.request.BrokerRequest
+import org.apache.pinot.common.utils.DataTable
+import org.apache.pinot.connector.spark.connector.Constants.PinotTableTypes
+import org.apache.pinot.connector.spark.datasource.PinotDataSourceReadOptions
+import org.apache.pinot.connector.spark.exceptions.PinotException
+import org.apache.pinot.connector.spark.utils.Logging
+import org.apache.pinot.core.transport.{AsyncQueryResponse, QueryRouter, ServerInstance}
+import org.apache.pinot.sql.parsers.CalciteSqlCompiler
+
+import scala.collection.JavaConverters._
+
+/**
+ * Fetch data from specified Pinot server.
+ */
+private[pinot] class PinotServerDataFetcher(
+ partitionId: Int,
+ pinotSplit: PinotSplit,
+ dataSourceOptions: PinotDataSourceReadOptions)
+ extends Logging {
+ private val sqlCompiler = new CalciteSqlCompiler()
Review comment:
SQL endpoint?
SQL is just used in query generation. User can create its sql query with lot of spark methods with filter push down etc. In the executor side, sql query is converted to the broker query. Then data will be received with `pinot.core.transport` APIs using broker request.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] fx19880617 merged pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
fx19880617 merged pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r464311850
##########
File path: config/.scalafmt.conf
##########
@@ -0,0 +1,17 @@
+version = "2.4.0"
Review comment:
Defined rules to format scala code.
This file removed due to travis fail. It required maven 3.6.0 version.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r463983154
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/PinotClusterClient.scala
##########
@@ -0,0 +1,208 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+import java.net.{URI, URLEncoder}
+import java.util.regex.Pattern
+
+import org.apache.pinot.connector.spark.decodeTo
+import org.apache.pinot.connector.spark.exceptions.{HttpStatusCodeException, PinotException}
+import org.apache.pinot.connector.spark.utils.{HttpUtils, Logging}
+
+import scala.util.{Failure, Success, Try}
+import io.circe.generic.auto._
+import org.apache.pinot.connector.spark.connector.Constants.PinotTableTypes
+import org.apache.pinot.connector.spark.connector.query.GeneratedPQLs
+import org.apache.pinot.spi.data.Schema
+
+/**
+ * Client that read/write/prepare required data from/to Pinot.
+ */
+private[pinot] object PinotClusterClient extends Logging {
+
+ def getTableSchema(controllerUrl: String, tableName: String): Schema = {
+ val rawTableName = PinotUtils.getRawTableName(tableName)
+ Try {
+ val uri = new URI(s"http://$controllerUrl/tables/$rawTableName/schema")
+ val response = HttpUtils.sendGetRequest(uri)
+ Schema.fromString(response)
+ } match {
+ case Success(response) =>
+ logDebug(s"Pinot schema received successfully for table '$rawTableName'")
+ response
+ case Failure(exception) =>
+ throw PinotException(
+ s"An error occurred while getting Pinot schema for table '$rawTableName'",
+ exception
+ )
+ }
+ }
+
+ /**
+ * Get available broker urls(host:port) for given table.
+ * This method is used when if broker instances not defined in the datasource options.
+ */
+ def getBrokerInstances(controllerUrl: String, tableName: String): List[String] = {
+ val brokerPattern = Pattern.compile("Broker_(.*)_(\\d+)")
+ val rawTableName = PinotUtils.getRawTableName(tableName)
+ Try {
+ val uri = new URI(s"http://$controllerUrl/tables/$rawTableName/instances")
Review comment:
What is the changes? How can i use?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r463983521
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/datasource/PinotDataSourceV2.scala
##########
@@ -0,0 +1,36 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.datasource
+
+import org.apache.spark.sql.sources.DataSourceRegister
+import org.apache.spark.sql.sources.v2.reader.DataSourceReader
+import org.apache.spark.sql.sources.v2.{DataSourceOptions, DataSourceV2, ReadSupport}
+import org.apache.spark.sql.types.StructType
+
+class PinotDataSourceV2 extends DataSourceV2 with ReadSupport with DataSourceRegister {
Review comment:
version2. In spark, generally, version is added as V1 or V2 to the datasource class to learn which datasource api was used
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#issuecomment-667574175
> Did the first pass and it looks very good. WIll be good to use SQL instead of PQL
PQL was changed with SQL.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r463983154
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/PinotClusterClient.scala
##########
@@ -0,0 +1,208 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+import java.net.{URI, URLEncoder}
+import java.util.regex.Pattern
+
+import org.apache.pinot.connector.spark.decodeTo
+import org.apache.pinot.connector.spark.exceptions.{HttpStatusCodeException, PinotException}
+import org.apache.pinot.connector.spark.utils.{HttpUtils, Logging}
+
+import scala.util.{Failure, Success, Try}
+import io.circe.generic.auto._
+import org.apache.pinot.connector.spark.connector.Constants.PinotTableTypes
+import org.apache.pinot.connector.spark.connector.query.GeneratedPQLs
+import org.apache.pinot.spi.data.Schema
+
+/**
+ * Client that read/write/prepare required data from/to Pinot.
+ */
+private[pinot] object PinotClusterClient extends Logging {
+
+ def getTableSchema(controllerUrl: String, tableName: String): Schema = {
+ val rawTableName = PinotUtils.getRawTableName(tableName)
+ Try {
+ val uri = new URI(s"http://$controllerUrl/tables/$rawTableName/schema")
+ val response = HttpUtils.sendGetRequest(uri)
+ Schema.fromString(response)
+ } match {
+ case Success(response) =>
+ logDebug(s"Pinot schema received successfully for table '$rawTableName'")
+ response
+ case Failure(exception) =>
+ throw PinotException(
+ s"An error occurred while getting Pinot schema for table '$rawTableName'",
+ exception
+ )
+ }
+ }
+
+ /**
+ * Get available broker urls(host:port) for given table.
+ * This method is used when if broker instances not defined in the datasource options.
+ */
+ def getBrokerInstances(controllerUrl: String, tableName: String): List[String] = {
+ val brokerPattern = Pattern.compile("Broker_(.*)_(\\d+)")
+ val rawTableName = PinotUtils.getRawTableName(tableName)
+ Try {
+ val uri = new URI(s"http://$controllerUrl/tables/$rawTableName/instances")
Review comment:
i'll check it. Connector was written with pinot 0.4.0. I can update it
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r464311850
##########
File path: config/.scalafmt.conf
##########
@@ -0,0 +1,17 @@
+version = "2.4.0"
Review comment:
Defined rules to format scala code
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r463983068
##########
File path: pinot-connectors/pinot-spark-connector/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
##########
@@ -0,0 +1 @@
+org.apache.pinot.connector.spark.datasource.PinotDataSourceV2
Review comment:
Spark has datasource read/write api v1 and v2. V2 api has support a lot of features like push down filters, columnar read etc. For more details [Spark DatasourceV2](https://www.slideshare.net/databricks/apache-spark-data-source-v2-with-wenchen-fan-and-gengliang-wang)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r468554247
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/PinotServerDataFetcher.scala
##########
@@ -0,0 +1,134 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+import java.util.{List => JList, Map => JMap}
+
+import com.yammer.metrics.core.MetricsRegistry
+import org.apache.helix.model.InstanceConfig
+import org.apache.pinot.common.metrics.BrokerMetrics
+import org.apache.pinot.common.request.BrokerRequest
+import org.apache.pinot.common.utils.DataTable
+import org.apache.pinot.connector.spark.connector.Constants.PinotTableTypes
+import org.apache.pinot.connector.spark.datasource.PinotDataSourceReadOptions
+import org.apache.pinot.connector.spark.exceptions.PinotException
+import org.apache.pinot.connector.spark.utils.Logging
+import org.apache.pinot.core.transport.{AsyncQueryResponse, QueryRouter, ServerInstance}
+import org.apache.pinot.sql.parsers.CalciteSqlCompiler
+
+import scala.collection.JavaConverters._
+
+/**
+ * Fetch data from specified Pinot server.
+ */
+private[pinot] class PinotServerDataFetcher(
+ partitionId: Int,
+ pinotSplit: PinotSplit,
+ dataSourceOptions: PinotDataSourceReadOptions)
+ extends Logging {
+ private val sqlCompiler = new CalciteSqlCompiler()
+ private val brokerId = "apache_spark"
Review comment:
I could not find any information about that for now. I'll look it in the next release
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r464513043
##########
File path: pinot-connectors/pinot-spark-connector/src/test/resources/schema/pinot-schema.json
##########
@@ -0,0 +1,57 @@
+{
Review comment:
yes
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#issuecomment-743471066
We will write the spark-pinot connector write section(segment creation from dataframe directly), then we are thinking that add streaming support. Now, we are designing write section
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#issuecomment-739889947
> hi @mangrrua
>
> "streaming endpoints" for spark connector currently this feature is available ?
No
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r464324725
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/query/SQLSelectionQueryGenerator.scala
##########
@@ -0,0 +1,101 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector.query
+
+import org.apache.pinot.connector.spark.connector.Constants.{PinotTableType, PinotTableTypes}
+import org.apache.pinot.connector.spark.connector.{PinotUtils, TimeBoundaryInfo}
+
+/**
+ * Generate realtime and offline SQL queries for specified table with given columns and filters.
+ */
+private[pinot] class SQLSelectionQueryGenerator(
+ tableNameWithType: String,
+ timeBoundaryInfo: Option[TimeBoundaryInfo],
+ columns: Array[String],
+ whereClause: Option[String]) {
+ private val columnsExpression = columnsAsExpression()
+ private val rawTableName = PinotUtils.getRawTableName(tableNameWithType)
+ private val tableType = PinotUtils.getTableType(tableNameWithType)
+
+ def generatePQLs(): GeneratedSQLs = {
+ val offlineSelectQuery = buildSelectQuery(PinotTableTypes.OFFLINE)
+ val realtimeSelectQuery = buildSelectQuery(PinotTableTypes.REALTIME)
+ GeneratedSQLs(
+ rawTableName,
+ tableType,
+ offlineSelectQuery,
+ realtimeSelectQuery
+ )
+ }
+
+ /**
+ * Get all columns if selecting columns empty(eg: resultDataFrame.count())
+ */
+ private def columnsAsExpression(): String = {
+ if (columns.isEmpty) "*" else columns.mkString(",")
+ }
+
+ /**
+ * Build realtime or offline PQL selection query.
+ */
+ private def buildSelectQuery(tableType: PinotTableType): String = {
+ val tableNameWithType = s"${rawTableName}_$tableType"
+ val queryBuilder = new StringBuilder(s"SELECT $columnsExpression FROM $tableNameWithType")
+
+ // add where clause if exists
+ whereClause.foreach { x =>
+ queryBuilder.append(s" WHERE $x")
+ }
+
+ // add time boundary filter if exists
+ timeBoundaryInfo.foreach { tbi =>
+ val timeBoundaryFilter =
+ if (tableType == PinotTableTypes.OFFLINE) {
+ tbi.getOfflinePredicate
+ } else {
+ tbi.getRealtimePredicate
+ }
+
+ if (whereClause.isEmpty) {
+ queryBuilder.append(s" WHERE $timeBoundaryFilter")
+ } else {
+ queryBuilder.append(s" AND $timeBoundaryFilter")
+ }
+ }
+
+ // query will be converted to Pinot 'BrokerRequest' with PQL compiler
+ // pinot set limit to 10 automatically
+ // to prevent this add limit to query
+ queryBuilder.append(s" LIMIT ${Int.MaxValue}")
Review comment:
I do not know. Connector supports only selection queries to read and re-index data etc. In this case, `LIMIT` is unnecessary i think except one case. If we will support `order by` in selection query, `limit` can be used. But this is not good practice while working with spark.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] fx19880617 commented on pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
fx19880617 commented on pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#issuecomment-741074495
+1 on the feature to write pinot segment directly from rdd or data frame!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#issuecomment-735817630
> @mangrrua any plans on implementing the write side of this? Seems very useful.
We are working on this.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] kishoreg commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
kishoreg commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r463982272
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/datasource/PinotDataSourceV2.scala
##########
@@ -0,0 +1,36 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.datasource
+
+import org.apache.spark.sql.sources.DataSourceRegister
+import org.apache.spark.sql.sources.v2.reader.DataSourceReader
+import org.apache.spark.sql.sources.v2.{DataSourceOptions, DataSourceV2, ReadSupport}
+import org.apache.spark.sql.types.StructType
+
+class PinotDataSourceV2 extends DataSourceV2 with ReadSupport with DataSourceRegister {
Review comment:
is there something special about v2 in spark? Trying to understand why we are naming PinotDataSourceV2 and not just PinotDataSource
##########
File path: pinot-connectors/pinot-spark-connector/pom.xml
##########
@@ -0,0 +1,246 @@
+<?xml version="1.0"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <artifactId>pinot-connectors</artifactId>
+ <groupId>org.apache.pinot</groupId>
+ <version>${revision}${sha1}</version>
+ <relativePath>..</relativePath>
+ </parent>
+ <artifactId>pinot-spark-connector</artifactId>
+ <name>Pinot Spark Connector</name>
+ <url>https://pinot.apache.org/</url>
+ <properties>
+ <pinot.root>${basedir}/../..</pinot.root>
+ <spark.version>2.4.5</spark.version>
+ <circe.version>0.13.0</circe.version>
+ </properties>
+
+ <profiles>
+ <profile>
+ <id>scala-2.12</id>
+ <activation>
+ <activeByDefault>true</activeByDefault>
+ </activation>
+ <properties>
+ <scala.version>2.12.11</scala.version>
+ <scala.compat.version>2.12</scala.compat.version>
+ </properties>
+ <dependencies>
+ <dependency>
+ <groupId>com.thoughtworks.paranamer</groupId>
+ <artifactId>paranamer</artifactId>
+ <version>2.8</version>
Review comment:
will be good to move all the versions to the properties
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/query/GeneratedPQLs.scala
##########
@@ -0,0 +1,33 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector.query
+
+import org.apache.pinot.connector.spark.connector.Constants.{PinotTableType, PinotTableTypes}
+
+private[pinot] case class GeneratedPQLs(
Review comment:
should we move to SQL since we plan to deprecate the PQL?
##########
File path: pinot-connectors/pinot-spark-connector/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
##########
@@ -0,0 +1 @@
+org.apache.pinot.connector.spark.datasource.PinotDataSourceV2
Review comment:
what is V2?
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/PinotClusterClient.scala
##########
@@ -0,0 +1,208 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+import java.net.{URI, URLEncoder}
+import java.util.regex.Pattern
+
+import org.apache.pinot.connector.spark.decodeTo
+import org.apache.pinot.connector.spark.exceptions.{HttpStatusCodeException, PinotException}
+import org.apache.pinot.connector.spark.utils.{HttpUtils, Logging}
+
+import scala.util.{Failure, Success, Try}
+import io.circe.generic.auto._
+import org.apache.pinot.connector.spark.connector.Constants.PinotTableTypes
+import org.apache.pinot.connector.spark.connector.query.GeneratedPQLs
+import org.apache.pinot.spi.data.Schema
+
+/**
+ * Client that read/write/prepare required data from/to Pinot.
+ */
+private[pinot] object PinotClusterClient extends Logging {
+
+ def getTableSchema(controllerUrl: String, tableName: String): Schema = {
+ val rawTableName = PinotUtils.getRawTableName(tableName)
+ Try {
+ val uri = new URI(s"http://$controllerUrl/tables/$rawTableName/schema")
+ val response = HttpUtils.sendGetRequest(uri)
+ Schema.fromString(response)
+ } match {
+ case Success(response) =>
+ logDebug(s"Pinot schema received successfully for table '$rawTableName'")
+ response
+ case Failure(exception) =>
+ throw PinotException(
+ s"An error occurred while getting Pinot schema for table '$rawTableName'",
+ exception
+ )
+ }
+ }
+
+ /**
+ * Get available broker urls(host:port) for given table.
+ * This method is used when if broker instances not defined in the datasource options.
+ */
+ def getBrokerInstances(controllerUrl: String, tableName: String): List[String] = {
+ val brokerPattern = Pattern.compile("Broker_(.*)_(\\d+)")
+ val rawTableName = PinotUtils.getRawTableName(tableName)
+ Try {
+ val uri = new URI(s"http://$controllerUrl/tables/$rawTableName/instances")
Review comment:
I think we have a new Controller API that provides this as one call cc @fx19880617
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r467567841
##########
File path: pinot-connectors/pinot-spark-connector/documentation/read_model.md
##########
@@ -0,0 +1,145 @@
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
+# Read Model
+
+Connector can scan offline, hybrid and realtime tables. `table` parameter have to given like below;
Review comment:
Hmm, table type option would be good. Table type option will be required option with the table name. It will be more clear, thanks!
Update: Implemented
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mayankshriv commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mayankshriv commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r464474155
##########
File path: pinot-connectors/pinot-spark-connector/src/test/resources/schema/pinot-schema.json
##########
@@ -0,0 +1,57 @@
+{
Review comment:
Perhaps we should have a PinotSchema <-> SparkSchema converter?
##########
File path: pinot-connectors/pinot-spark-connector/src/test/scala/org/apache/pinot/connector/spark/connector/PinotSplitterTest.scala
##########
@@ -0,0 +1,95 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+import org.apache.pinot.connector.spark.BaseTest
+import org.apache.pinot.connector.spark.connector.Constants.PinotTableTypes
+import org.apache.pinot.connector.spark.connector.query.GeneratedSQLs
+import org.apache.pinot.connector.spark.exceptions.PinotException
+
+class PinotSplitterTest extends BaseTest {
Review comment:
Could we have docs on all classes?
##########
File path: pinot-connectors/pinot-spark-connector/documentation/read_model.md
##########
@@ -0,0 +1,145 @@
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
+# Read Model
Review comment:
Should this also go into https://docs.pinot.apache.org, after the PR is committed?
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/datasource/PinotDataSourceV2.scala
##########
@@ -0,0 +1,36 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.datasource
+
+import org.apache.spark.sql.sources.DataSourceRegister
+import org.apache.spark.sql.sources.v2.reader.DataSourceReader
+import org.apache.spark.sql.sources.v2.{DataSourceOptions, DataSourceV2, ReadSupport}
+import org.apache.spark.sql.types.StructType
+
+class PinotDataSourceV2 extends DataSourceV2 with ReadSupport with DataSourceRegister {
Review comment:
Adding docs will help, for folks not yet hands on with Scala.
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/datasource/PinotDataSourceReader.scala
##########
@@ -0,0 +1,124 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.datasource
+
+import java.util.{List => JList}
+
+import org.apache.pinot.connector.spark.connector.query.SQLSelectionQueryGenerator
+import org.apache.pinot.connector.spark.connector.{
+ FilterPushDown,
+ PinotClusterClient,
+ PinotSplitter,
+ PinotUtils
+}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.sources._
+import org.apache.spark.sql.sources.v2.DataSourceOptions
+import org.apache.spark.sql.sources.v2.reader.{
+ DataSourceReader,
+ InputPartition,
+ SupportsPushDownFilters,
+ SupportsPushDownRequiredColumns
+}
+import org.apache.spark.sql.types._
+
+import scala.collection.JavaConverters._
+
+class PinotDataSourceReader(options: DataSourceOptions, userSchema: Option[StructType] = None)
+ extends DataSourceReader
+ with SupportsPushDownFilters
+ with SupportsPushDownRequiredColumns {
+
+ private val pinotDataSourceOptions = PinotDataSourceReadOptions.from(options)
+ private var acceptedFilters: Array[Filter] = Array.empty
+ private var currentSchema: StructType = _
+
+ override def readSchema(): StructType = {
+ if (currentSchema == null) {
+ currentSchema = userSchema.getOrElse {
+ val pinotTableSchema = PinotClusterClient.getTableSchema(
+ pinotDataSourceOptions.controller,
+ pinotDataSourceOptions.tableName
+ )
+ PinotUtils.pinotSchemaToSparkSchema(pinotTableSchema)
+ }
+ }
+ currentSchema
+ }
+
+ override def planInputPartitions(): JList[InputPartition[InternalRow]] = {
+ val schema = readSchema()
+ val tableType = PinotUtils.getTableType(pinotDataSourceOptions.tableName)
+
+ // Time boundary is used when table is hybrid to ensure that the overlap
+ // between realtime and offline segment data is queried exactly once
+ val timeBoundaryInfo =
+ if (tableType.isDefined) {
+ None
+ } else {
+ PinotClusterClient.getTimeBoundaryInfo(
+ pinotDataSourceOptions.broker,
+ pinotDataSourceOptions.tableName
+ )
+ }
+
+ val whereCondition = FilterPushDown.compileFiltersToSqlWhereClause(this.acceptedFilters)
+ val generatedSQLs = SQLSelectionQueryGenerator.generate(
+ pinotDataSourceOptions.tableName,
+ timeBoundaryInfo,
+ schema.fieldNames,
+ whereCondition
+ )
+
+ val routingTable =
+ PinotClusterClient.getRoutingTable(pinotDataSourceOptions.broker, generatedSQLs)
Review comment:
Connecting to pinot server directly leads to querying routing-table / time-boundary, which the broker does. Wondering if there is plan to connect via the broker to avoid this? It may have the following advantages:
* No need to query routing-table / time-boundary, and unlike in this approach.
* Filter push down
One issue I see though, it may not be feasible to stream data out of broker with the current code. I am trying to see what the general direction/approach is with these connectors.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r464543147
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/PinotServerDataFetcher.scala
##########
@@ -0,0 +1,134 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+import java.util.{List => JList, Map => JMap}
+
+import com.yammer.metrics.core.MetricsRegistry
+import org.apache.helix.model.InstanceConfig
+import org.apache.pinot.common.metrics.BrokerMetrics
+import org.apache.pinot.common.request.BrokerRequest
+import org.apache.pinot.common.utils.DataTable
+import org.apache.pinot.connector.spark.connector.Constants.PinotTableTypes
+import org.apache.pinot.connector.spark.datasource.PinotDataSourceReadOptions
+import org.apache.pinot.connector.spark.exceptions.PinotException
+import org.apache.pinot.connector.spark.utils.Logging
+import org.apache.pinot.core.transport.{AsyncQueryResponse, QueryRouter, ServerInstance}
+import org.apache.pinot.sql.parsers.CalciteSqlCompiler
+
+import scala.collection.JavaConverters._
+
+/**
+ * Fetch data from specified Pinot server.
+ */
+private[pinot] class PinotServerDataFetcher(
+ partitionId: Int,
+ pinotSplit: PinotSplit,
+ dataSourceOptions: PinotDataSourceReadOptions)
+ extends Logging {
+ private val sqlCompiler = new CalciteSqlCompiler()
Review comment:
I've created PR related this, can you check? @fx19880617
https://github.com/apache/incubator-pinot/pull/5791
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r463983521
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/datasource/PinotDataSourceV2.scala
##########
@@ -0,0 +1,36 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.datasource
+
+import org.apache.spark.sql.sources.DataSourceRegister
+import org.apache.spark.sql.sources.v2.reader.DataSourceReader
+import org.apache.spark.sql.sources.v2.{DataSourceOptions, DataSourceV2, ReadSupport}
+import org.apache.spark.sql.types.StructType
+
+class PinotDataSourceV2 extends DataSourceV2 with ReadSupport with DataSourceRegister {
Review comment:
version2. In spark, generally, version is added as V1 or V2 to the datasource class.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] flykent1990 commented on pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
flykent1990 commented on pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#issuecomment-740379158
@mangrrua New version for spark connector supports write when publish ? This feature is very useful for pinot! I look forward to :)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r464522910
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/PinotClusterClient.scala
##########
@@ -0,0 +1,208 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+import java.net.{URI, URLEncoder}
+import java.util.regex.Pattern
+
+import org.apache.pinot.connector.spark.decodeTo
+import org.apache.pinot.connector.spark.exceptions.{HttpStatusCodeException, PinotException}
+import org.apache.pinot.connector.spark.utils.{HttpUtils, Logging}
+
+import scala.util.{Failure, Success, Try}
+import io.circe.generic.auto._
+import org.apache.pinot.connector.spark.connector.Constants.PinotTableTypes
+import org.apache.pinot.connector.spark.connector.query.GeneratedSQLs
+import org.apache.pinot.spi.data.Schema
+
+/**
+ * Client that read/write/prepare required data from/to Pinot.
+ */
+private[pinot] object PinotClusterClient extends Logging {
+
+ def getTableSchema(controllerUrl: String, tableName: String): Schema = {
+ val rawTableName = PinotUtils.getRawTableName(tableName)
+ Try {
+ val uri = new URI(s"http://$controllerUrl/tables/$rawTableName/schema")
+ val response = HttpUtils.sendGetRequest(uri)
+ Schema.fromString(response)
Review comment:
Yes, but i guess required one more http request? is this way outdated? @fx19880617
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r463983285
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/query/GeneratedPQLs.scala
##########
@@ -0,0 +1,33 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector.query
+
+import org.apache.pinot.connector.spark.connector.Constants.{PinotTableType, PinotTableTypes}
+
+private[pinot] case class GeneratedPQLs(
Review comment:
we can convert pql to sql. i'll update supported filters and query generation for that, and use sql compiler instead of pql-compiler
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r464317075
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/PinotSplitter.scala
##########
@@ -0,0 +1,85 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+import java.util.regex.{Matcher, Pattern}
+
+import org.apache.pinot.connector.spark.connector.Constants.PinotTableType
+import org.apache.pinot.connector.spark.connector.query.GeneratedSQLs
+import org.apache.pinot.connector.spark.exceptions.PinotException
+import org.apache.pinot.connector.spark.utils.Logging
+
+private[pinot] object PinotSplitter extends Logging {
+ private val PINOT_SERVER_PATTERN = Pattern.compile("Server_(.*)_(\\d+)")
+
+ def generatePinotSplits(
+ generatedSQLs: GeneratedSQLs,
+ routingTable: Map[String, Map[String, List[String]]],
+ segmentsPerSplit: Int): List[PinotSplit] = {
+ routingTable.flatMap {
+ case (tableType, serversToSegments) =>
+ serversToSegments
+ .map { case (server, segments) => parseServerInput(server, segments) }
+ .flatMap {
+ case (matcher, segments) =>
+ createPinotSplitsFromSubSplits(
+ tableType,
+ generatedSQLs,
+ matcher,
+ segments,
+ segmentsPerSplit
+ )
+ }
+ }.toList
+ }
+
+ private def parseServerInput(server: String, segments: List[String]): (Matcher, List[String]) = {
+ val matcher = PINOT_SERVER_PATTERN.matcher(server)
+ if (matcher.matches() && matcher.groupCount() == 2) matcher -> segments
+ else throw PinotException(s"'$server' did not match!?")
+ }
+
+ private def createPinotSplitsFromSubSplits(
+ tableType: PinotTableType,
+ generatedSQLs: GeneratedSQLs,
+ serverMatcher: Matcher,
+ segments: List[String],
+ segmentsPerSplit: Int): Iterator[PinotSplit] = {
+ val serverHost = serverMatcher.group(1)
+ val serverPort = serverMatcher.group(2)
+ val maxSegmentCount = Math.min(segments.size, segmentsPerSplit)
+ segments.grouped(maxSegmentCount).map { subSegments =>
+ val serverAndSegments =
+ PinotServerAndSegments(serverHost, serverPort, subSegments, tableType)
+ PinotSplit(generatedSQLs, serverAndSegments)
+ }
+ }
+}
+
+private[pinot] case class PinotSplit(
+ generatedSQLs: GeneratedSQLs,
+ serverAndSegments: PinotServerAndSegments)
+
+private[pinot] case class PinotServerAndSegments(
+ serverHost: String,
+ serverPort: String,
+ segments: List[String],
+ serverType: PinotTableType) {
Review comment:
`GeneratedSQLs` contains the realtime and offline queries. `PinotSplit` contains the generated queries with the server and segment informations. In the executor side, we are send query to the specific server. Connector checks the `serverType` to learn which query will be converted, and sent to the which server. Thus we need to the `serverType` in the current design. This is the usage;
```scala
val pinotServerAsyncQueryResponse = pinotSplit.serverAndSegments.serverType match {
case TableType.REALTIME =>
val realtimeBrokerRequest =
sqlCompiler.compileToBrokerRequest(pinotSplit.generatedSQLs.realtimeSelectQuery)
submitRequestToPinotServer(null, null, realtimeBrokerRequest, routingTableForRequest)
case TableType.OFFLINE =>
val offlineBrokerRequest =
sqlCompiler.compileToBrokerRequest(pinotSplit.generatedSQLs.offlineSelectQuery)
submitRequestToPinotServer(offlineBrokerRequest, routingTableForRequest, null, null)
}
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r464524102
##########
File path: pinot-connectors/pinot-spark-connector/documentation/read_model.md
##########
@@ -0,0 +1,145 @@
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
+# Read Model
+
+Connector can scan offline, hybrid and realtime tables. `table` parameter have to given like below;
+- For offline table `tbl_OFFLINE`
+- For realtime table `tbl_REALTIME`
+- For hybrid table `tbl`
+
+An example scan;
+
+```scala
+val df = spark.read
+ .format("pinot")
+ .option("table", "airlineStats")
+ .load()
+```
+
+Custom schema can be specified directly. If schema is not specified, connector read table schema from Pinot controller, and then convert to the Spark schema.
+
+
+### Architecture
+
+Connector reads data from `Pinot Servers` directly. For this operation, firstly, connector creates query with given filters(if filter push down is enabled) and columns, then finds routing table for created query. It creates pinot splits that contains **ONE PINOT SERVER and ONE OR MORE SEGMENT per spark partition**, based on the routing table and `segmentsPerSplit`(detailed explain is defined below). Lastly, each partition read data from specified pinot server in parallel.
+
+
+
+
+Each Spark partition open connection with Pinot server, and read data. For example, assume that routing table informations for specified query is like that:
+
+```
+- realtime ->
+ - realtimeServer1 -> (segment1, segment2, segment3)
+ - realtimeServer2 -> (segment4)
+- offline ->
+ - offlineServer10 -> (segment10, segment20)
+```
+
+If `segmentsPerSplit` is equal to 3, there will be created 3 Spark partition like below;
+
+| Spark Partition | Queried Pinot Server/Segments |
+| ------------- | ------------- |
+| partition1 | realtimeServer1 / segment1, segment2, segment3 |
+| partition2 | realtimeServer2 / segment4 |
+| partition3 | offlineServer10 / segment10, segment20 |
+
+
+If `segmentsPerSplit` is equal to 1, there will be created 6 Spark partition;
+
+| Spark Partition | Queried Pinot Server/Segments |
+| ------------- | ------------- |
+| partition1 | realtimeServer1 / segment1 |
+| partition2 | realtimeServer1 / segment2 |
+| partition3 | realtimeServer1 / segment3 |
+| partition4 | realtimeServer2 / segment4 |
+| partition5 | offlineServer10 / segment10 |
+| partition6 | offlineServer10 / segment20 |
+
+
+If `segmentsPerSplit` value is too low, that means more parallelism. But this also mean that a lot of connection will be opened with Pinot servers, and will increase QPS on the Pinot servers.
+
+If `segmetnsPerSplit` value is too high, that means less parallelism. Each Pinot server will scan more segments per request.
+
+**Note:** Pinot servers prunes segments based on the segment metadata when query comes. In some cases(for example filtering based on the some columns), some servers may not return data. Therefore, some Spark partitions will be empty. In this cases, `repartition()` may be applied for efficient data analysis after loading data to Spark.
+
+
+### Filter And Column Push Down
+Connector supports filter and column push down. Filters and columns are pushed to the pinot servers. Filter and column push down improves the performance while reading data because of its minimizing data transfer between Pinot and Spark. In default, filter push down enabled. If filters are desired to be applied in Spark, `usePushDownFilters` should be set as `false`.
+
+Connector supports `Equal, In, LessThan, LessThanOrEqual, Greater, GreaterThan, Not, TEXT_MATCH, And, Or` filters for now.
Review comment:
The filters section of the readme outdated. I've changed pql with sql, but i did forget to change supported filters section. The connector supports all sql filters now. I'll fix it
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r467567841
##########
File path: pinot-connectors/pinot-spark-connector/documentation/read_model.md
##########
@@ -0,0 +1,145 @@
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
+# Read Model
+
+Connector can scan offline, hybrid and realtime tables. `table` parameter have to given like below;
Review comment:
Hmm, table type option would be good. Table type option will be required option with the table name. It will be more clear, thanks!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] codecov-commenter commented on pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#issuecomment-671048574
# [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/5787?src=pr&el=h1) Report
> Merging [#5787](https://codecov.io/gh/apache/incubator-pinot/pull/5787?src=pr&el=desc) into [master](https://codecov.io/gh/apache/incubator-pinot/commit/1beaab59b73f26c4e35f3b9bc856b03806cddf5a&el=desc) will **decrease** coverage by `21.84%`.
> The diff coverage is `54.29%`.
[](https://codecov.io/gh/apache/incubator-pinot/pull/5787?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #5787 +/- ##
===========================================
- Coverage 66.44% 44.60% -21.85%
===========================================
Files 1075 1144 +69
Lines 54773 59370 +4597
Branches 8168 9089 +921
===========================================
- Hits 36396 26483 -9913
- Misses 15700 30484 +14784
+ Partials 2677 2403 -274
```
| Flag | Coverage Δ | |
|---|---|---|
| #integration | `44.60% <54.29%> (?)` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/incubator-pinot/pull/5787?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [...ot/broker/broker/AllowAllAccessControlFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/5787/diff?src=pr&el=tree#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvYnJva2VyL0FsbG93QWxsQWNjZXNzQ29udHJvbEZhY3RvcnkuamF2YQ==) | `100.00% <ø> (ø)` | |
| [...ava/org/apache/pinot/client/AbstractResultSet.java](https://codecov.io/gh/apache/incubator-pinot/pull/5787/diff?src=pr&el=tree#diff-cGlub3QtY2xpZW50cy9waW5vdC1qYXZhLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY2xpZW50L0Fic3RyYWN0UmVzdWx0U2V0LmphdmE=) | `26.66% <0.00%> (-30.48%)` | :arrow_down: |
| [.../main/java/org/apache/pinot/client/Connection.java](https://codecov.io/gh/apache/incubator-pinot/pull/5787/diff?src=pr&el=tree#diff-cGlub3QtY2xpZW50cy9waW5vdC1qYXZhLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY2xpZW50L0Nvbm5lY3Rpb24uamF2YQ==) | `22.22% <0.00%> (-26.62%)` | :arrow_down: |
| [.../org/apache/pinot/client/ResultTableResultSet.java](https://codecov.io/gh/apache/incubator-pinot/pull/5787/diff?src=pr&el=tree#diff-cGlub3QtY2xpZW50cy9waW5vdC1qYXZhLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY2xpZW50L1Jlc3VsdFRhYmxlUmVzdWx0U2V0LmphdmE=) | `24.00% <0.00%> (-10.29%)` | :arrow_down: |
| [.../org/apache/pinot/common/lineage/LineageEntry.java](https://codecov.io/gh/apache/incubator-pinot/pull/5787/diff?src=pr&el=tree#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbGluZWFnZS9MaW5lYWdlRW50cnkuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...apache/pinot/common/lineage/LineageEntryState.java](https://codecov.io/gh/apache/incubator-pinot/pull/5787/diff?src=pr&el=tree#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbGluZWFnZS9MaW5lYWdlRW50cnlTdGF0ZS5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...rg/apache/pinot/common/lineage/SegmentLineage.java](https://codecov.io/gh/apache/incubator-pinot/pull/5787/diff?src=pr&el=tree#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbGluZWFnZS9TZWdtZW50TGluZWFnZS5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...ache/pinot/common/lineage/SegmentLineageUtils.java](https://codecov.io/gh/apache/incubator-pinot/pull/5787/diff?src=pr&el=tree#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbGluZWFnZS9TZWdtZW50TGluZWFnZVV0aWxzLmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...restlet/resources/StartReplaceSegmentsRequest.java](https://codecov.io/gh/apache/incubator-pinot/pull/5787/diff?src=pr&el=tree#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vcmVzdGxldC9yZXNvdXJjZXMvU3RhcnRSZXBsYWNlU2VnbWVudHNSZXF1ZXN0LmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...java/org/apache/pinot/common/segment/ReadMode.java](https://codecov.io/gh/apache/incubator-pinot/pull/5787/diff?src=pr&el=tree#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vc2VnbWVudC9SZWFkTW9kZS5qYXZh) | `66.66% <ø> (ø)` | |
| ... and [1062 more](https://codecov.io/gh/apache/incubator-pinot/pull/5787/diff?src=pr&el=tree-more) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/5787?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/5787?src=pr&el=footer). Last update [373ceb1...2388d96](https://codecov.io/gh/apache/incubator-pinot/pull/5787?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] tannghia025 commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
tannghia025 commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r568417075
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/datasource/PinotDataSourceReader.scala
##########
@@ -0,0 +1,124 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.datasource
+
+import java.util.{List => JList}
+
+import org.apache.pinot.connector.spark.connector.query.SQLSelectionQueryGenerator
+import org.apache.pinot.connector.spark.connector.{
+ FilterPushDown,
+ PinotClusterClient,
+ PinotSplitter,
+ PinotUtils
+}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.sources._
+import org.apache.spark.sql.sources.v2.DataSourceOptions
+import org.apache.spark.sql.sources.v2.reader.{
+ DataSourceReader,
+ InputPartition,
+ SupportsPushDownFilters,
+ SupportsPushDownRequiredColumns
+}
+import org.apache.spark.sql.types._
+
+import scala.collection.JavaConverters._
+
+class PinotDataSourceReader(options: DataSourceOptions, userSchema: Option[StructType] = None)
+ extends DataSourceReader
+ with SupportsPushDownFilters
+ with SupportsPushDownRequiredColumns {
+
+ private val pinotDataSourceOptions = PinotDataSourceReadOptions.from(options)
+ private var acceptedFilters: Array[Filter] = Array.empty
+ private var currentSchema: StructType = _
+
+ override def readSchema(): StructType = {
+ if (currentSchema == null) {
+ currentSchema = userSchema.getOrElse {
+ val pinotTableSchema = PinotClusterClient.getTableSchema(
+ pinotDataSourceOptions.controller,
+ pinotDataSourceOptions.tableName
+ )
+ PinotUtils.pinotSchemaToSparkSchema(pinotTableSchema)
+ }
+ }
+ currentSchema
+ }
+
+ override def planInputPartitions(): JList[InputPartition[InternalRow]] = {
+ val schema = readSchema()
+ val tableType = PinotUtils.getTableType(pinotDataSourceOptions.tableName)
+
+ // Time boundary is used when table is hybrid to ensure that the overlap
+ // between realtime and offline segment data is queried exactly once
+ val timeBoundaryInfo =
+ if (tableType.isDefined) {
+ None
+ } else {
+ PinotClusterClient.getTimeBoundaryInfo(
+ pinotDataSourceOptions.broker,
+ pinotDataSourceOptions.tableName
+ )
+ }
+
+ val whereCondition = FilterPushDown.compileFiltersToSqlWhereClause(this.acceptedFilters)
+ val generatedSQLs = SQLSelectionQueryGenerator.generate(
+ pinotDataSourceOptions.tableName,
+ timeBoundaryInfo,
+ schema.fieldNames,
+ whereCondition
+ )
+
+ val routingTable =
+ PinotClusterClient.getRoutingTable(pinotDataSourceOptions.broker, generatedSQLs)
Review comment:
ok
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] quanghaisoft commented on pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
quanghaisoft commented on pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#issuecomment-759181651
> We will write the spark-pinot connector write section(segment creation from dataframe directly), then we are thinking that add streaming support. Now, we are designing write section
Please help me upgrade this lib to spark 3.0.x , scala 2.12.x thank you very much
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r464543147
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/PinotServerDataFetcher.scala
##########
@@ -0,0 +1,134 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+import java.util.{List => JList, Map => JMap}
+
+import com.yammer.metrics.core.MetricsRegistry
+import org.apache.helix.model.InstanceConfig
+import org.apache.pinot.common.metrics.BrokerMetrics
+import org.apache.pinot.common.request.BrokerRequest
+import org.apache.pinot.common.utils.DataTable
+import org.apache.pinot.connector.spark.connector.Constants.PinotTableTypes
+import org.apache.pinot.connector.spark.datasource.PinotDataSourceReadOptions
+import org.apache.pinot.connector.spark.exceptions.PinotException
+import org.apache.pinot.connector.spark.utils.Logging
+import org.apache.pinot.core.transport.{AsyncQueryResponse, QueryRouter, ServerInstance}
+import org.apache.pinot.sql.parsers.CalciteSqlCompiler
+
+import scala.collection.JavaConverters._
+
+/**
+ * Fetch data from specified Pinot server.
+ */
+private[pinot] class PinotServerDataFetcher(
+ partitionId: Int,
+ pinotSplit: PinotSplit,
+ dataSourceOptions: PinotDataSourceReadOptions)
+ extends Logging {
+ private val sqlCompiler = new CalciteSqlCompiler()
Review comment:
I've created PR related this, can you check? @fx19880617
https://github.com/apache/incubator-pinot/pull/5791
I've changed routing table endpoint(in `PinotClusterClient`) with new endpoint in PR #5791
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r464504242
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/datasource/PinotDataSourceReader.scala
##########
@@ -0,0 +1,124 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.datasource
+
+import java.util.{List => JList}
+
+import org.apache.pinot.connector.spark.connector.query.SQLSelectionQueryGenerator
+import org.apache.pinot.connector.spark.connector.{
+ FilterPushDown,
+ PinotClusterClient,
+ PinotSplitter,
+ PinotUtils
+}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.sources._
+import org.apache.spark.sql.sources.v2.DataSourceOptions
+import org.apache.spark.sql.sources.v2.reader.{
+ DataSourceReader,
+ InputPartition,
+ SupportsPushDownFilters,
+ SupportsPushDownRequiredColumns
+}
+import org.apache.spark.sql.types._
+
+import scala.collection.JavaConverters._
+
+class PinotDataSourceReader(options: DataSourceOptions, userSchema: Option[StructType] = None)
+ extends DataSourceReader
+ with SupportsPushDownFilters
+ with SupportsPushDownRequiredColumns {
+
+ private val pinotDataSourceOptions = PinotDataSourceReadOptions.from(options)
+ private var acceptedFilters: Array[Filter] = Array.empty
+ private var currentSchema: StructType = _
+
+ override def readSchema(): StructType = {
+ if (currentSchema == null) {
+ currentSchema = userSchema.getOrElse {
+ val pinotTableSchema = PinotClusterClient.getTableSchema(
+ pinotDataSourceOptions.controller,
+ pinotDataSourceOptions.tableName
+ )
+ PinotUtils.pinotSchemaToSparkSchema(pinotTableSchema)
+ }
+ }
+ currentSchema
+ }
+
+ override def planInputPartitions(): JList[InputPartition[InternalRow]] = {
+ val schema = readSchema()
+ val tableType = PinotUtils.getTableType(pinotDataSourceOptions.tableName)
+
+ // Time boundary is used when table is hybrid to ensure that the overlap
+ // between realtime and offline segment data is queried exactly once
+ val timeBoundaryInfo =
+ if (tableType.isDefined) {
+ None
+ } else {
+ PinotClusterClient.getTimeBoundaryInfo(
+ pinotDataSourceOptions.broker,
+ pinotDataSourceOptions.tableName
+ )
+ }
+
+ val whereCondition = FilterPushDown.compileFiltersToSqlWhereClause(this.acceptedFilters)
+ val generatedSQLs = SQLSelectionQueryGenerator.generate(
+ pinotDataSourceOptions.tableName,
+ timeBoundaryInfo,
+ schema.fieldNames,
+ whereCondition
+ )
+
+ val routingTable =
+ PinotClusterClient.getRoutingTable(pinotDataSourceOptions.broker, generatedSQLs)
Review comment:
Routing table is required. Spark reads data in parallel like broker. Each server-segments represented as spark partition. This is most efficient way to receive data from Pinot. Also Presto uses this architecture in pinot-connector.
If broker not available, connector throws exception. To avoid this, we can create a related API in controller side, and use it like metadata manager in other systems.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r464326792
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/PinotServerDataFetcher.scala
##########
@@ -0,0 +1,134 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+import java.util.{List => JList, Map => JMap}
+
+import com.yammer.metrics.core.MetricsRegistry
+import org.apache.helix.model.InstanceConfig
+import org.apache.pinot.common.metrics.BrokerMetrics
+import org.apache.pinot.common.request.BrokerRequest
+import org.apache.pinot.common.utils.DataTable
+import org.apache.pinot.connector.spark.connector.Constants.PinotTableTypes
+import org.apache.pinot.connector.spark.datasource.PinotDataSourceReadOptions
+import org.apache.pinot.connector.spark.exceptions.PinotException
+import org.apache.pinot.connector.spark.utils.Logging
+import org.apache.pinot.core.transport.{AsyncQueryResponse, QueryRouter, ServerInstance}
+import org.apache.pinot.sql.parsers.CalciteSqlCompiler
+
+import scala.collection.JavaConverters._
+
+/**
+ * Fetch data from specified Pinot server.
+ */
+private[pinot] class PinotServerDataFetcher(
+ partitionId: Int,
+ pinotSplit: PinotSplit,
+ dataSourceOptions: PinotDataSourceReadOptions)
+ extends Logging {
+ private val sqlCompiler = new CalciteSqlCompiler()
Review comment:
Yeah. I catch one thing. Connector uses broker `debug/routingTable?pql` API to get routing table for a query(for segment pruning). But in connector, i've changed pql to sql. I'll add sql endpoint to get routing table.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r464543147
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/PinotServerDataFetcher.scala
##########
@@ -0,0 +1,134 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+import java.util.{List => JList, Map => JMap}
+
+import com.yammer.metrics.core.MetricsRegistry
+import org.apache.helix.model.InstanceConfig
+import org.apache.pinot.common.metrics.BrokerMetrics
+import org.apache.pinot.common.request.BrokerRequest
+import org.apache.pinot.common.utils.DataTable
+import org.apache.pinot.connector.spark.connector.Constants.PinotTableTypes
+import org.apache.pinot.connector.spark.datasource.PinotDataSourceReadOptions
+import org.apache.pinot.connector.spark.exceptions.PinotException
+import org.apache.pinot.connector.spark.utils.Logging
+import org.apache.pinot.core.transport.{AsyncQueryResponse, QueryRouter, ServerInstance}
+import org.apache.pinot.sql.parsers.CalciteSqlCompiler
+
+import scala.collection.JavaConverters._
+
+/**
+ * Fetch data from specified Pinot server.
+ */
+private[pinot] class PinotServerDataFetcher(
+ partitionId: Int,
+ pinotSplit: PinotSplit,
+ dataSourceOptions: PinotDataSourceReadOptions)
+ extends Logging {
+ private val sqlCompiler = new CalciteSqlCompiler()
Review comment:
I've created PR related this, can you check? @fx19880617
https://github.com/apache/incubator-pinot/pull/5791
I've changed routing table endpoint with new endpoint in PR #5791
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r463983068
##########
File path: pinot-connectors/pinot-spark-connector/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
##########
@@ -0,0 +1 @@
+org.apache.pinot.connector.spark.datasource.PinotDataSourceV2
Review comment:
version.
Spark has datasource read/write api v1 and v2. V2 api has support a lot of features like push down filters, columnar read etc. For more details [Spark DatasourceV2](https://www.slideshare.net/databricks/apache-spark-data-source-v2-with-wenchen-fan-and-gengliang-wang)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r463983068
##########
File path: pinot-connectors/pinot-spark-connector/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
##########
@@ -0,0 +1 @@
+org.apache.pinot.connector.spark.datasource.PinotDataSourceV2
Review comment:
Spark has datasource read/write api v1 and v2. V2 api has support a lot of features like push down filters, columnar read etc. For for details [Spark DatasourceV2](https://www.slideshare.net/databricks/apache-spark-data-source-v2-with-wenchen-fan-and-gengliang-wang)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] tannghia025 commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
tannghia025 commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r568417075
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/datasource/PinotDataSourceReader.scala
##########
@@ -0,0 +1,124 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.datasource
+
+import java.util.{List => JList}
+
+import org.apache.pinot.connector.spark.connector.query.SQLSelectionQueryGenerator
+import org.apache.pinot.connector.spark.connector.{
+ FilterPushDown,
+ PinotClusterClient,
+ PinotSplitter,
+ PinotUtils
+}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.sources._
+import org.apache.spark.sql.sources.v2.DataSourceOptions
+import org.apache.spark.sql.sources.v2.reader.{
+ DataSourceReader,
+ InputPartition,
+ SupportsPushDownFilters,
+ SupportsPushDownRequiredColumns
+}
+import org.apache.spark.sql.types._
+
+import scala.collection.JavaConverters._
+
+class PinotDataSourceReader(options: DataSourceOptions, userSchema: Option[StructType] = None)
+ extends DataSourceReader
+ with SupportsPushDownFilters
+ with SupportsPushDownRequiredColumns {
+
+ private val pinotDataSourceOptions = PinotDataSourceReadOptions.from(options)
+ private var acceptedFilters: Array[Filter] = Array.empty
+ private var currentSchema: StructType = _
+
+ override def readSchema(): StructType = {
+ if (currentSchema == null) {
+ currentSchema = userSchema.getOrElse {
+ val pinotTableSchema = PinotClusterClient.getTableSchema(
+ pinotDataSourceOptions.controller,
+ pinotDataSourceOptions.tableName
+ )
+ PinotUtils.pinotSchemaToSparkSchema(pinotTableSchema)
+ }
+ }
+ currentSchema
+ }
+
+ override def planInputPartitions(): JList[InputPartition[InternalRow]] = {
+ val schema = readSchema()
+ val tableType = PinotUtils.getTableType(pinotDataSourceOptions.tableName)
+
+ // Time boundary is used when table is hybrid to ensure that the overlap
+ // between realtime and offline segment data is queried exactly once
+ val timeBoundaryInfo =
+ if (tableType.isDefined) {
+ None
+ } else {
+ PinotClusterClient.getTimeBoundaryInfo(
+ pinotDataSourceOptions.broker,
+ pinotDataSourceOptions.tableName
+ )
+ }
+
+ val whereCondition = FilterPushDown.compileFiltersToSqlWhereClause(this.acceptedFilters)
+ val generatedSQLs = SQLSelectionQueryGenerator.generate(
+ pinotDataSourceOptions.tableName,
+ timeBoundaryInfo,
+ schema.fieldNames,
+ whereCondition
+ )
+
+ val routingTable =
+ PinotClusterClient.getRoutingTable(pinotDataSourceOptions.broker, generatedSQLs)
Review comment:
ok
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] noahprince22 commented on pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
noahprince22 commented on pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#issuecomment-726326612
@mangrrua any plans on implementing the write side of this? Seems very useful.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] fx19880617 commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
fx19880617 commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r468275673
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/datasource/PinotDataSourceReader.scala
##########
@@ -0,0 +1,124 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.datasource
+
+import java.util.{List => JList}
+
+import org.apache.pinot.connector.spark.connector.query.SQLSelectionQueryGenerator
+import org.apache.pinot.connector.spark.connector.{
+ FilterPushDown,
+ PinotClusterClient,
+ PinotSplitter,
+ PinotUtils
+}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.sources._
+import org.apache.spark.sql.sources.v2.DataSourceOptions
+import org.apache.spark.sql.sources.v2.reader.{
+ DataSourceReader,
+ InputPartition,
+ SupportsPushDownFilters,
+ SupportsPushDownRequiredColumns
+}
+import org.apache.spark.sql.types._
+
+import scala.collection.JavaConverters._
+
+class PinotDataSourceReader(options: DataSourceOptions, userSchema: Option[StructType] = None)
+ extends DataSourceReader
+ with SupportsPushDownFilters
+ with SupportsPushDownRequiredColumns {
+
+ private val pinotDataSourceOptions = PinotDataSourceReadOptions.from(options)
+ private var acceptedFilters: Array[Filter] = Array.empty
+ private var currentSchema: StructType = _
+
+ override def readSchema(): StructType = {
+ if (currentSchema == null) {
+ currentSchema = userSchema.getOrElse {
+ val pinotTableSchema = PinotClusterClient.getTableSchema(
+ pinotDataSourceOptions.controller,
+ pinotDataSourceOptions.tableName
+ )
+ PinotUtils.pinotSchemaToSparkSchema(pinotTableSchema)
+ }
+ }
+ currentSchema
+ }
+
+ override def planInputPartitions(): JList[InputPartition[InternalRow]] = {
+ val schema = readSchema()
+ val tableType = PinotUtils.getTableType(pinotDataSourceOptions.tableName)
+
+ // Time boundary is used when table is hybrid to ensure that the overlap
+ // between realtime and offline segment data is queried exactly once
+ val timeBoundaryInfo =
+ if (tableType.isDefined) {
+ None
+ } else {
+ PinotClusterClient.getTimeBoundaryInfo(
+ pinotDataSourceOptions.broker,
+ pinotDataSourceOptions.tableName
+ )
+ }
+
+ val whereCondition = FilterPushDown.compileFiltersToSqlWhereClause(this.acceptedFilters)
+ val generatedSQLs = SQLSelectionQueryGenerator.generate(
+ pinotDataSourceOptions.tableName,
+ timeBoundaryInfo,
+ schema.fieldNames,
+ whereCondition
+ )
+
+ val routingTable =
+ PinotClusterClient.getRoutingTable(pinotDataSourceOptions.broker, generatedSQLs)
Review comment:
> Connecting to pinot server directly leads to querying routing-table / time-boundary, which the broker does. Wondering if there is plan to connect via the broker to avoid this? It may have the following advantages:
>
> * No need to query routing-table / time-boundary, and unlike in this approach.
> * Filter push down
>
> One issue I see though, it may not be feasible to stream data out of broker with the current code. I am trying to see what the general direction/approach is with these connectors.
We are opening the gate on server with streaming api which can be used by presto for more scalable solution. This is in general the same usage pattern: make pinot data queryable in warehouse.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#issuecomment-667951838
> This is great work! Looking forward to streaming API for read and writeAPI :)
>
> One question maybe related to this, shall we create a PinotRDD?
Thank you :)
Spark DatasourceV2 API supports very flexibility and features. We do not need to create an RDD. We are working on the dataframe and dataset api. After read data from pinot, all you have to the is play in spark..
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r464522910
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/PinotClusterClient.scala
##########
@@ -0,0 +1,208 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+import java.net.{URI, URLEncoder}
+import java.util.regex.Pattern
+
+import org.apache.pinot.connector.spark.decodeTo
+import org.apache.pinot.connector.spark.exceptions.{HttpStatusCodeException, PinotException}
+import org.apache.pinot.connector.spark.utils.{HttpUtils, Logging}
+
+import scala.util.{Failure, Success, Try}
+import io.circe.generic.auto._
+import org.apache.pinot.connector.spark.connector.Constants.PinotTableTypes
+import org.apache.pinot.connector.spark.connector.query.GeneratedSQLs
+import org.apache.pinot.spi.data.Schema
+
+/**
+ * Client that read/write/prepare required data from/to Pinot.
+ */
+private[pinot] object PinotClusterClient extends Logging {
+
+ def getTableSchema(controllerUrl: String, tableName: String): Schema = {
+ val rawTableName = PinotUtils.getRawTableName(tableName)
+ Try {
+ val uri = new URI(s"http://$controllerUrl/tables/$rawTableName/schema")
+ val response = HttpUtils.sendGetRequest(uri)
+ Schema.fromString(response)
Review comment:
Yes, but i guess required one more http request? is this way is outdated? @fx19880617
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r464526782
##########
File path: pinot-connectors/pinot-spark-connector/documentation/read_model.md
##########
@@ -0,0 +1,145 @@
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
+# Read Model
+
+Connector can scan offline, hybrid and realtime tables. `table` parameter have to given like below;
Review comment:
Users can not specify table name without type if table not hybrid. If table hybrid, it checks routing tables of realtime and offline query separately.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] fx19880617 commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
fx19880617 commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r467512585
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/query/SQLSelectionQueryGenerator.scala
##########
@@ -0,0 +1,101 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector.query
+
+import org.apache.pinot.connector.spark.connector.Constants.{PinotTableType, PinotTableTypes}
+import org.apache.pinot.connector.spark.connector.{PinotUtils, TimeBoundaryInfo}
+
+/**
+ * Generate realtime and offline SQL queries for specified table with given columns and filters.
+ */
+private[pinot] class SQLSelectionQueryGenerator(
+ tableNameWithType: String,
+ timeBoundaryInfo: Option[TimeBoundaryInfo],
+ columns: Array[String],
+ whereClause: Option[String]) {
+ private val columnsExpression = columnsAsExpression()
+ private val rawTableName = PinotUtils.getRawTableName(tableNameWithType)
+ private val tableType = PinotUtils.getTableType(tableNameWithType)
+
+ def generatePQLs(): GeneratedSQLs = {
+ val offlineSelectQuery = buildSelectQuery(PinotTableTypes.OFFLINE)
+ val realtimeSelectQuery = buildSelectQuery(PinotTableTypes.REALTIME)
+ GeneratedSQLs(
+ rawTableName,
+ tableType,
+ offlineSelectQuery,
+ realtimeSelectQuery
+ )
+ }
+
+ /**
+ * Get all columns if selecting columns empty(eg: resultDataFrame.count())
+ */
+ private def columnsAsExpression(): String = {
+ if (columns.isEmpty) "*" else columns.mkString(",")
+ }
+
+ /**
+ * Build realtime or offline PQL selection query.
+ */
+ private def buildSelectQuery(tableType: PinotTableType): String = {
+ val tableNameWithType = s"${rawTableName}_$tableType"
+ val queryBuilder = new StringBuilder(s"SELECT $columnsExpression FROM $tableNameWithType")
+
+ // add where clause if exists
+ whereClause.foreach { x =>
+ queryBuilder.append(s" WHERE $x")
+ }
+
+ // add time boundary filter if exists
+ timeBoundaryInfo.foreach { tbi =>
+ val timeBoundaryFilter =
+ if (tableType == PinotTableTypes.OFFLINE) {
+ tbi.getOfflinePredicate
+ } else {
+ tbi.getRealtimePredicate
+ }
+
+ if (whereClause.isEmpty) {
+ queryBuilder.append(s" WHERE $timeBoundaryFilter")
+ } else {
+ queryBuilder.append(s" AND $timeBoundaryFilter")
+ }
+ }
+
+ // query will be converted to Pinot 'BrokerRequest' with PQL compiler
+ // pinot set limit to 10 automatically
+ // to prevent this add limit to query
+ queryBuilder.append(s" LIMIT ${Int.MaxValue}")
Review comment:
Got it.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r464544165
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/PinotServerDataFetcher.scala
##########
@@ -0,0 +1,134 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+import java.util.{List => JList, Map => JMap}
+
+import com.yammer.metrics.core.MetricsRegistry
+import org.apache.helix.model.InstanceConfig
+import org.apache.pinot.common.metrics.BrokerMetrics
+import org.apache.pinot.common.request.BrokerRequest
+import org.apache.pinot.common.utils.DataTable
+import org.apache.pinot.connector.spark.connector.Constants.PinotTableTypes
+import org.apache.pinot.connector.spark.datasource.PinotDataSourceReadOptions
+import org.apache.pinot.connector.spark.exceptions.PinotException
+import org.apache.pinot.connector.spark.utils.Logging
+import org.apache.pinot.core.transport.{AsyncQueryResponse, QueryRouter, ServerInstance}
+import org.apache.pinot.sql.parsers.CalciteSqlCompiler
+
+import scala.collection.JavaConverters._
+
+/**
+ * Fetch data from specified Pinot server.
+ */
+private[pinot] class PinotServerDataFetcher(
+ partitionId: Int,
+ pinotSplit: PinotSplit,
+ dataSourceOptions: PinotDataSourceReadOptions)
+ extends Logging {
+ private val sqlCompiler = new CalciteSqlCompiler()
+ private val brokerId = "apache_spark"
Review comment:
i'll look for more info about executor
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] fx19880617 commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
fx19880617 commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r464071943
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/PinotClusterClient.scala
##########
@@ -0,0 +1,208 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+import java.net.{URI, URLEncoder}
+import java.util.regex.Pattern
+
+import org.apache.pinot.connector.spark.decodeTo
+import org.apache.pinot.connector.spark.exceptions.{HttpStatusCodeException, PinotException}
+import org.apache.pinot.connector.spark.utils.{HttpUtils, Logging}
+
+import scala.util.{Failure, Success, Try}
+import io.circe.generic.auto._
+import org.apache.pinot.connector.spark.connector.Constants.PinotTableTypes
+import org.apache.pinot.connector.spark.connector.query.GeneratedSQLs
+import org.apache.pinot.spi.data.Schema
+
+/**
+ * Client that read/write/prepare required data from/to Pinot.
+ */
+private[pinot] object PinotClusterClient extends Logging {
+
+ def getTableSchema(controllerUrl: String, tableName: String): Schema = {
+ val rawTableName = PinotUtils.getRawTableName(tableName)
+ Try {
+ val uri = new URI(s"http://$controllerUrl/tables/$rawTableName/schema")
+ val response = HttpUtils.sendGetRequest(uri)
+ Schema.fromString(response)
+ } match {
+ case Success(response) =>
+ logDebug(s"Pinot schema received successfully for table '$rawTableName'")
+ response
+ case Failure(exception) =>
+ throw PinotException(
+ s"An error occurred while getting Pinot schema for table '$rawTableName'",
+ exception
+ )
+ }
+ }
+
+ /**
+ * Get available broker urls(host:port) for given table.
+ * This method is used when if broker instances not defined in the datasource options.
+ */
+ def getBrokerInstances(controllerUrl: String, tableName: String): List[String] = {
+ val brokerPattern = Pattern.compile("Broker_(.*)_(\\d+)")
+ val rawTableName = PinotUtils.getRawTableName(tableName)
+ Try {
+ val uri = new URI(s"http://$controllerUrl/tables/$rawTableName/instances")
+ val response = HttpUtils.sendGetRequest(uri)
+ val brokerUrls = decodeTo[PinotInstances](response).brokers
+ .flatMap(_.instances)
+ .distinct
+ .map(brokerPattern.matcher)
+ .filter(matcher => matcher.matches() && matcher.groupCount() == 2)
+ .map { matcher =>
+ val host = matcher.group(1)
+ val port = matcher.group(2)
+ s"$host:$port"
+ }
+
+ if (brokerUrls.isEmpty) {
+ throw new IllegalStateException(s"Not found broker instance for table '$rawTableName'")
+ }
+
+ brokerUrls
+ } match {
+ case Success(result) =>
+ logDebug(s"Broker instances received successfully for table '$tableName'")
+ result
+ case Failure(exception) =>
+ throw PinotException(
+ s"An error occurred while getting broker instances for table '$rawTableName'",
+ exception
+ )
+ }
+ }
+
+ /**
+ * Get time boundary info of specified table.
+ * This method is used when table is hybrid to ensure that the overlap
+ * between realtime and offline segment data is queried exactly once.
+ *
+ * @return time boundary info if table exist and segments push type is 'append' or None otherwise
+ */
+ def getTimeBoundaryInfo(brokerUrl: String, tableName: String): Option[TimeBoundaryInfo] = {
+ val rawTableName = PinotUtils.getRawTableName(tableName)
+ Try {
+ // pinot converts the given table name to the offline table name automatically
+ val uri = new URI(s"http://$brokerUrl/debug/timeBoundary/$rawTableName")
Review comment:
Suggest to make all those URI patterns to be constant.
##########
File path: config/.scalafmt.conf
##########
@@ -0,0 +1,17 @@
+version = "2.4.0"
Review comment:
What's this file for?
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/PinotServerDataFetcher.scala
##########
@@ -0,0 +1,134 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+import java.util.{List => JList, Map => JMap}
+
+import com.yammer.metrics.core.MetricsRegistry
+import org.apache.helix.model.InstanceConfig
+import org.apache.pinot.common.metrics.BrokerMetrics
+import org.apache.pinot.common.request.BrokerRequest
+import org.apache.pinot.common.utils.DataTable
+import org.apache.pinot.connector.spark.connector.Constants.PinotTableTypes
+import org.apache.pinot.connector.spark.datasource.PinotDataSourceReadOptions
+import org.apache.pinot.connector.spark.exceptions.PinotException
+import org.apache.pinot.connector.spark.utils.Logging
+import org.apache.pinot.core.transport.{AsyncQueryResponse, QueryRouter, ServerInstance}
+import org.apache.pinot.sql.parsers.CalciteSqlCompiler
+
+import scala.collection.JavaConverters._
+
+/**
+ * Fetch data from specified Pinot server.
+ */
+private[pinot] class PinotServerDataFetcher(
+ partitionId: Int,
+ pinotSplit: PinotSplit,
+ dataSourceOptions: PinotDataSourceReadOptions)
+ extends Logging {
+ private val sqlCompiler = new CalciteSqlCompiler()
+ private val brokerId = "apache_spark"
Review comment:
possible to get more granular info about the executor?
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/PinotServerDataFetcher.scala
##########
@@ -0,0 +1,134 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+import java.util.{List => JList, Map => JMap}
+
+import com.yammer.metrics.core.MetricsRegistry
+import org.apache.helix.model.InstanceConfig
+import org.apache.pinot.common.metrics.BrokerMetrics
+import org.apache.pinot.common.request.BrokerRequest
+import org.apache.pinot.common.utils.DataTable
+import org.apache.pinot.connector.spark.connector.Constants.PinotTableTypes
+import org.apache.pinot.connector.spark.datasource.PinotDataSourceReadOptions
+import org.apache.pinot.connector.spark.exceptions.PinotException
+import org.apache.pinot.connector.spark.utils.Logging
+import org.apache.pinot.core.transport.{AsyncQueryResponse, QueryRouter, ServerInstance}
+import org.apache.pinot.sql.parsers.CalciteSqlCompiler
+
+import scala.collection.JavaConverters._
+
+/**
+ * Fetch data from specified Pinot server.
+ */
+private[pinot] class PinotServerDataFetcher(
+ partitionId: Int,
+ pinotSplit: PinotSplit,
+ dataSourceOptions: PinotDataSourceReadOptions)
+ extends Logging {
+ private val sqlCompiler = new CalciteSqlCompiler()
Review comment:
Since we are using CalciteSqlCompiler here, I assume we are always using new SQL endpoint, could you change all the `PQL` references to `SQL`?
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/query/SQLSelectionQueryGenerator.scala
##########
@@ -0,0 +1,101 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector.query
+
+import org.apache.pinot.connector.spark.connector.Constants.{PinotTableType, PinotTableTypes}
+import org.apache.pinot.connector.spark.connector.{PinotUtils, TimeBoundaryInfo}
+
+/**
+ * Generate realtime and offline SQL queries for specified table with given columns and filters.
+ */
+private[pinot] class SQLSelectionQueryGenerator(
+ tableNameWithType: String,
+ timeBoundaryInfo: Option[TimeBoundaryInfo],
+ columns: Array[String],
+ whereClause: Option[String]) {
+ private val columnsExpression = columnsAsExpression()
+ private val rawTableName = PinotUtils.getRawTableName(tableNameWithType)
+ private val tableType = PinotUtils.getTableType(tableNameWithType)
+
+ def generatePQLs(): GeneratedSQLs = {
Review comment:
`generateSQLs`?
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/query/SQLSelectionQueryGenerator.scala
##########
@@ -0,0 +1,101 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector.query
+
+import org.apache.pinot.connector.spark.connector.Constants.{PinotTableType, PinotTableTypes}
+import org.apache.pinot.connector.spark.connector.{PinotUtils, TimeBoundaryInfo}
+
+/**
+ * Generate realtime and offline SQL queries for specified table with given columns and filters.
+ */
+private[pinot] class SQLSelectionQueryGenerator(
+ tableNameWithType: String,
+ timeBoundaryInfo: Option[TimeBoundaryInfo],
+ columns: Array[String],
+ whereClause: Option[String]) {
+ private val columnsExpression = columnsAsExpression()
+ private val rawTableName = PinotUtils.getRawTableName(tableNameWithType)
+ private val tableType = PinotUtils.getTableType(tableNameWithType)
+
+ def generatePQLs(): GeneratedSQLs = {
+ val offlineSelectQuery = buildSelectQuery(PinotTableTypes.OFFLINE)
+ val realtimeSelectQuery = buildSelectQuery(PinotTableTypes.REALTIME)
+ GeneratedSQLs(
+ rawTableName,
+ tableType,
+ offlineSelectQuery,
+ realtimeSelectQuery
+ )
+ }
+
+ /**
+ * Get all columns if selecting columns empty(eg: resultDataFrame.count())
+ */
+ private def columnsAsExpression(): String = {
+ if (columns.isEmpty) "*" else columns.mkString(",")
+ }
+
+ /**
+ * Build realtime or offline PQL selection query.
+ */
+ private def buildSelectQuery(tableType: PinotTableType): String = {
+ val tableNameWithType = s"${rawTableName}_$tableType"
+ val queryBuilder = new StringBuilder(s"SELECT $columnsExpression FROM $tableNameWithType")
+
+ // add where clause if exists
+ whereClause.foreach { x =>
+ queryBuilder.append(s" WHERE $x")
+ }
+
+ // add time boundary filter if exists
+ timeBoundaryInfo.foreach { tbi =>
+ val timeBoundaryFilter =
+ if (tableType == PinotTableTypes.OFFLINE) {
+ tbi.getOfflinePredicate
+ } else {
+ tbi.getRealtimePredicate
+ }
+
+ if (whereClause.isEmpty) {
+ queryBuilder.append(s" WHERE $timeBoundaryFilter")
+ } else {
+ queryBuilder.append(s" AND $timeBoundaryFilter")
+ }
+ }
+
+ // query will be converted to Pinot 'BrokerRequest' with PQL compiler
+ // pinot set limit to 10 automatically
+ // to prevent this add limit to query
+ queryBuilder.append(s" LIMIT ${Int.MaxValue}")
Review comment:
shall we check if there is existing `LIMIT`?
##########
File path: pinot-connectors/pinot-spark-connector/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
##########
@@ -0,0 +1 @@
+org.apache.pinot.connector.spark.datasource.PinotDataSourceV2
Review comment:
new line
##########
File path: pinot-connectors/pinot-spark-connector/documentation/read_model.md
##########
@@ -0,0 +1,145 @@
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
+# Read Model
+
+Connector can scan offline, hybrid and realtime tables. `table` parameter have to given like below;
Review comment:
Worth to mention that if the table is OFFLINE or REALTIME only table, user can also use `tbl` to query.
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/Constants.scala
##########
@@ -0,0 +1,28 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+private[pinot] object Constants {
+ type PinotTableType = String
+
+ object PinotTableTypes {
Review comment:
`pinot-spi` has already defined enum: `org.apache.pinot.spi.config.table.TableType`
Also you may want to check some utils for table name: `org.apache.pinot.spi.utils.builder.TableNameBuilder`
##########
File path: pinot-connectors/pinot-spark-connector/documentation/read_model.md
##########
@@ -0,0 +1,145 @@
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
+# Read Model
+
+Connector can scan offline, hybrid and realtime tables. `table` parameter have to given like below;
+- For offline table `tbl_OFFLINE`
+- For realtime table `tbl_REALTIME`
+- For hybrid table `tbl`
+
+An example scan;
+
+```scala
+val df = spark.read
+ .format("pinot")
+ .option("table", "airlineStats")
+ .load()
+```
+
+Custom schema can be specified directly. If schema is not specified, connector read table schema from Pinot controller, and then convert to the Spark schema.
+
+
+### Architecture
+
+Connector reads data from `Pinot Servers` directly. For this operation, firstly, connector creates query with given filters(if filter push down is enabled) and columns, then finds routing table for created query. It creates pinot splits that contains **ONE PINOT SERVER and ONE OR MORE SEGMENT per spark partition**, based on the routing table and `segmentsPerSplit`(detailed explain is defined below). Lastly, each partition read data from specified pinot server in parallel.
+
+
+
+
+Each Spark partition open connection with Pinot server, and read data. For example, assume that routing table informations for specified query is like that:
+
+```
+- realtime ->
+ - realtimeServer1 -> (segment1, segment2, segment3)
+ - realtimeServer2 -> (segment4)
+- offline ->
+ - offlineServer10 -> (segment10, segment20)
+```
+
+If `segmentsPerSplit` is equal to 3, there will be created 3 Spark partition like below;
+
+| Spark Partition | Queried Pinot Server/Segments |
+| ------------- | ------------- |
+| partition1 | realtimeServer1 / segment1, segment2, segment3 |
+| partition2 | realtimeServer2 / segment4 |
+| partition3 | offlineServer10 / segment10, segment20 |
+
+
+If `segmentsPerSplit` is equal to 1, there will be created 6 Spark partition;
+
+| Spark Partition | Queried Pinot Server/Segments |
+| ------------- | ------------- |
+| partition1 | realtimeServer1 / segment1 |
+| partition2 | realtimeServer1 / segment2 |
+| partition3 | realtimeServer1 / segment3 |
+| partition4 | realtimeServer2 / segment4 |
+| partition5 | offlineServer10 / segment10 |
+| partition6 | offlineServer10 / segment20 |
+
+
+If `segmentsPerSplit` value is too low, that means more parallelism. But this also mean that a lot of connection will be opened with Pinot servers, and will increase QPS on the Pinot servers.
+
+If `segmetnsPerSplit` value is too high, that means less parallelism. Each Pinot server will scan more segments per request.
+
+**Note:** Pinot servers prunes segments based on the segment metadata when query comes. In some cases(for example filtering based on the some columns), some servers may not return data. Therefore, some Spark partitions will be empty. In this cases, `repartition()` may be applied for efficient data analysis after loading data to Spark.
+
+
+### Filter And Column Push Down
+Connector supports filter and column push down. Filters and columns are pushed to the pinot servers. Filter and column push down improves the performance while reading data because of its minimizing data transfer between Pinot and Spark. In default, filter push down enabled. If filters are desired to be applied in Spark, `usePushDownFilters` should be set as `false`.
+
+Connector supports `Equal, In, LessThan, LessThanOrEqual, Greater, GreaterThan, Not, TEXT_MATCH, And, Or` filters for now.
Review comment:
For filter push down do we support just column with filters like (`columnA > 5`) or we can support more like transform function on columns like (`columnA > columnB`, `columnA * 10 > columnB`)
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/PinotClusterClient.scala
##########
@@ -0,0 +1,208 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+import java.net.{URI, URLEncoder}
+import java.util.regex.Pattern
+
+import org.apache.pinot.connector.spark.decodeTo
+import org.apache.pinot.connector.spark.exceptions.{HttpStatusCodeException, PinotException}
+import org.apache.pinot.connector.spark.utils.{HttpUtils, Logging}
+
+import scala.util.{Failure, Success, Try}
+import io.circe.generic.auto._
+import org.apache.pinot.connector.spark.connector.Constants.PinotTableTypes
+import org.apache.pinot.connector.spark.connector.query.GeneratedSQLs
+import org.apache.pinot.spi.data.Schema
+
+/**
+ * Client that read/write/prepare required data from/to Pinot.
+ */
+private[pinot] object PinotClusterClient extends Logging {
+
+ def getTableSchema(controllerUrl: String, tableName: String): Schema = {
+ val rawTableName = PinotUtils.getRawTableName(tableName)
+ Try {
+ val uri = new URI(s"http://$controllerUrl/tables/$rawTableName/schema")
+ val response = HttpUtils.sendGetRequest(uri)
+ Schema.fromString(response)
Review comment:
If you can get schema name from the table config, then you can use method `getSchema(host, port, schema)` in `org.apache.pinot.common.utils.SchemaUtils`
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/PinotUtils.scala
##########
@@ -0,0 +1,140 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+import org.apache.pinot.common.utils.DataSchema.ColumnDataType
+import org.apache.pinot.common.utils.DataTable
+import org.apache.pinot.connector.spark.connector.Constants.{PinotTableType, PinotTableTypes}
+import org.apache.pinot.connector.spark.exceptions.PinotException
+import org.apache.pinot.spi.data.{FieldSpec, Schema}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.ArrayData
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+import scala.collection.JavaConverters._
+
+private[pinot] object PinotUtils {
+ private val OFFLINE_TABLE_SUFFIX = s"_${PinotTableTypes.OFFLINE}"
+ private val REALTIME_TABLE_SUFFIX = s"_${PinotTableTypes.REALTIME}"
+
+ /** Extract raw pinot table name. */
+ def getRawTableName(tableName: String): String = {
+ if (tableName.endsWith(OFFLINE_TABLE_SUFFIX)) {
+ tableName.substring(0, tableName.length - OFFLINE_TABLE_SUFFIX.length)
+ } else if (tableName.endsWith(REALTIME_TABLE_SUFFIX)) {
+ tableName.substring(0, tableName.length - REALTIME_TABLE_SUFFIX.length)
+ } else {
+ tableName
+ }
+ }
+
+ /** Return offline/realtime table type, or None if table is hybrid. */
+ def getTableType(tableName: String): Option[PinotTableType] = {
+ if (tableName.endsWith(OFFLINE_TABLE_SUFFIX)) {
+ Some(PinotTableTypes.OFFLINE)
+ } else if (tableName.endsWith(REALTIME_TABLE_SUFFIX)) {
+ Some(PinotTableTypes.REALTIME)
+ } else {
+ None
+ }
+ }
+
+ /** Convert a Pinot schema to Spark schema. */
+ def pinotSchemaToSparkSchema(schema: Schema): StructType = {
+ val structFields = schema.getAllFieldSpecs.asScala.map { field =>
+ val sparkDataType = pinotDataTypeToSparkDataType(field.getDataType)
+ if (field.isSingleValueField) {
+ StructField(field.getName, sparkDataType)
+ } else {
+ StructField(field.getName, ArrayType(sparkDataType))
+ }
+ }
+ StructType(structFields.toList)
+ }
+
+ private def pinotDataTypeToSparkDataType(dataType: FieldSpec.DataType): DataType =
+ dataType match {
+ case FieldSpec.DataType.INT => IntegerType
+ case FieldSpec.DataType.LONG => LongType
+ case FieldSpec.DataType.FLOAT => FloatType
+ case FieldSpec.DataType.DOUBLE => DoubleType
+ case FieldSpec.DataType.STRING => StringType
+ case _ =>
+ throw PinotException(s"Unsupported pinot data type '$dataType")
+ }
+
+ /** Convert Pinot DataTable to Seq of InternalRow */
+ def pinotDataTableToInternalRows(
+ dataTable: DataTable,
+ sparkSchema: StructType): Seq[InternalRow] = {
+ val dataTableColumnNames = dataTable.getDataSchema.getColumnNames
+ (0 until dataTable.getNumberOfRows).map { rowIndex =>
+ // spark schema is used to ensure columns order
+ val columns = sparkSchema.fields.map { field =>
+ val colIndex = dataTableColumnNames.indexOf(field.name)
+ if (colIndex < 0) {
+ throw PinotException(s"'${field.name}' not found in Pinot server response")
+ } else {
+ // pinot column data type can be used directly,
+ // because all of them is supported in spark schema
+ val columnDataType = dataTable.getDataSchema.getColumnDataType(colIndex)
+ readPinotColumnData(dataTable, columnDataType, rowIndex, colIndex)
+ }
+ }
+ InternalRow.fromSeq(columns)
+ }
+ }
+
+ private def readPinotColumnData(
+ dataTable: DataTable,
+ columnDataType: ColumnDataType,
+ rowIndex: Int,
+ colIndex: Int): Any = columnDataType match {
+ // single column types
+ case ColumnDataType.STRING =>
Review comment:
what about `BYTES`?
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/PinotUtils.scala
##########
@@ -0,0 +1,140 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+import org.apache.pinot.common.utils.DataSchema.ColumnDataType
+import org.apache.pinot.common.utils.DataTable
+import org.apache.pinot.connector.spark.connector.Constants.{PinotTableType, PinotTableTypes}
+import org.apache.pinot.connector.spark.exceptions.PinotException
+import org.apache.pinot.spi.data.{FieldSpec, Schema}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.ArrayData
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+import scala.collection.JavaConverters._
+
+private[pinot] object PinotUtils {
+ private val OFFLINE_TABLE_SUFFIX = s"_${PinotTableTypes.OFFLINE}"
+ private val REALTIME_TABLE_SUFFIX = s"_${PinotTableTypes.REALTIME}"
+
+ /** Extract raw pinot table name. */
+ def getRawTableName(tableName: String): String = {
+ if (tableName.endsWith(OFFLINE_TABLE_SUFFIX)) {
Review comment:
Try to see if you can reuse the methods in `org.apache.pinot.spi.utils.builder.TableNameBuilder`
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/PinotUtils.scala
##########
@@ -0,0 +1,140 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+import org.apache.pinot.common.utils.DataSchema.ColumnDataType
+import org.apache.pinot.common.utils.DataTable
+import org.apache.pinot.connector.spark.connector.Constants.{PinotTableType, PinotTableTypes}
+import org.apache.pinot.connector.spark.exceptions.PinotException
+import org.apache.pinot.spi.data.{FieldSpec, Schema}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.ArrayData
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+import scala.collection.JavaConverters._
+
+private[pinot] object PinotUtils {
+ private val OFFLINE_TABLE_SUFFIX = s"_${PinotTableTypes.OFFLINE}"
+ private val REALTIME_TABLE_SUFFIX = s"_${PinotTableTypes.REALTIME}"
+
+ /** Extract raw pinot table name. */
+ def getRawTableName(tableName: String): String = {
+ if (tableName.endsWith(OFFLINE_TABLE_SUFFIX)) {
+ tableName.substring(0, tableName.length - OFFLINE_TABLE_SUFFIX.length)
+ } else if (tableName.endsWith(REALTIME_TABLE_SUFFIX)) {
+ tableName.substring(0, tableName.length - REALTIME_TABLE_SUFFIX.length)
+ } else {
+ tableName
+ }
+ }
+
+ /** Return offline/realtime table type, or None if table is hybrid. */
+ def getTableType(tableName: String): Option[PinotTableType] = {
Review comment:
`TableNameBuilder .getTableTypeFromTableName()`
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/PinotSplitter.scala
##########
@@ -0,0 +1,85 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+import java.util.regex.{Matcher, Pattern}
+
+import org.apache.pinot.connector.spark.connector.Constants.PinotTableType
+import org.apache.pinot.connector.spark.connector.query.GeneratedSQLs
+import org.apache.pinot.connector.spark.exceptions.PinotException
+import org.apache.pinot.connector.spark.utils.Logging
+
+private[pinot] object PinotSplitter extends Logging {
+ private val PINOT_SERVER_PATTERN = Pattern.compile("Server_(.*)_(\\d+)")
+
+ def generatePinotSplits(
+ generatedSQLs: GeneratedSQLs,
+ routingTable: Map[String, Map[String, List[String]]],
+ segmentsPerSplit: Int): List[PinotSplit] = {
+ routingTable.flatMap {
+ case (tableType, serversToSegments) =>
+ serversToSegments
+ .map { case (server, segments) => parseServerInput(server, segments) }
+ .flatMap {
+ case (matcher, segments) =>
+ createPinotSplitsFromSubSplits(
+ tableType,
+ generatedSQLs,
+ matcher,
+ segments,
+ segmentsPerSplit
+ )
+ }
+ }.toList
+ }
+
+ private def parseServerInput(server: String, segments: List[String]): (Matcher, List[String]) = {
+ val matcher = PINOT_SERVER_PATTERN.matcher(server)
+ if (matcher.matches() && matcher.groupCount() == 2) matcher -> segments
+ else throw PinotException(s"'$server' did not match!?")
+ }
+
+ private def createPinotSplitsFromSubSplits(
+ tableType: PinotTableType,
+ generatedSQLs: GeneratedSQLs,
+ serverMatcher: Matcher,
+ segments: List[String],
+ segmentsPerSplit: Int): Iterator[PinotSplit] = {
+ val serverHost = serverMatcher.group(1)
+ val serverPort = serverMatcher.group(2)
+ val maxSegmentCount = Math.min(segments.size, segmentsPerSplit)
+ segments.grouped(maxSegmentCount).map { subSegments =>
+ val serverAndSegments =
+ PinotServerAndSegments(serverHost, serverPort, subSegments, tableType)
+ PinotSplit(generatedSQLs, serverAndSegments)
+ }
+ }
+}
+
+private[pinot] case class PinotSplit(
+ generatedSQLs: GeneratedSQLs,
+ serverAndSegments: PinotServerAndSegments)
+
+private[pinot] case class PinotServerAndSegments(
+ serverHost: String,
+ serverPort: String,
+ segments: List[String],
+ serverType: PinotTableType) {
Review comment:
Why we need this ? Ideally it should just be inside the `GeneratedSQLs`? So we will send generated query to the given server for some segments.
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/PinotClusterClient.scala
##########
@@ -0,0 +1,208 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+import java.net.{URI, URLEncoder}
+import java.util.regex.Pattern
+
+import org.apache.pinot.connector.spark.decodeTo
+import org.apache.pinot.connector.spark.exceptions.{HttpStatusCodeException, PinotException}
+import org.apache.pinot.connector.spark.utils.{HttpUtils, Logging}
+
+import scala.util.{Failure, Success, Try}
+import io.circe.generic.auto._
+import org.apache.pinot.connector.spark.connector.Constants.PinotTableTypes
+import org.apache.pinot.connector.spark.connector.query.GeneratedPQLs
+import org.apache.pinot.spi.data.Schema
+
+/**
+ * Client that read/write/prepare required data from/to Pinot.
+ */
+private[pinot] object PinotClusterClient extends Logging {
+
+ def getTableSchema(controllerUrl: String, tableName: String): Schema = {
+ val rawTableName = PinotUtils.getRawTableName(tableName)
+ Try {
+ val uri = new URI(s"http://$controllerUrl/tables/$rawTableName/schema")
+ val response = HttpUtils.sendGetRequest(uri)
+ Schema.fromString(response)
+ } match {
+ case Success(response) =>
+ logDebug(s"Pinot schema received successfully for table '$rawTableName'")
+ response
+ case Failure(exception) =>
+ throw PinotException(
+ s"An error occurred while getting Pinot schema for table '$rawTableName'",
+ exception
+ )
+ }
+ }
+
+ /**
+ * Get available broker urls(host:port) for given table.
+ * This method is used when if broker instances not defined in the datasource options.
+ */
+ def getBrokerInstances(controllerUrl: String, tableName: String): List[String] = {
+ val brokerPattern = Pattern.compile("Broker_(.*)_(\\d+)")
+ val rawTableName = PinotUtils.getRawTableName(tableName)
+ Try {
+ val uri = new URI(s"http://$controllerUrl/tables/$rawTableName/instances")
Review comment:
here: https://github.com/apache/incubator-pinot/pull/5685
##########
File path: pinot-connectors/pinot-spark-connector/src/main/scala/org/apache/pinot/connector/spark/connector/PinotUtils.scala
##########
@@ -0,0 +1,140 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.connector.spark.connector
+
+import org.apache.pinot.common.utils.DataSchema.ColumnDataType
+import org.apache.pinot.common.utils.DataTable
+import org.apache.pinot.connector.spark.connector.Constants.{PinotTableType, PinotTableTypes}
+import org.apache.pinot.connector.spark.exceptions.PinotException
+import org.apache.pinot.spi.data.{FieldSpec, Schema}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.ArrayData
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+import scala.collection.JavaConverters._
+
+private[pinot] object PinotUtils {
+ private val OFFLINE_TABLE_SUFFIX = s"_${PinotTableTypes.OFFLINE}"
+ private val REALTIME_TABLE_SUFFIX = s"_${PinotTableTypes.REALTIME}"
+
+ /** Extract raw pinot table name. */
+ def getRawTableName(tableName: String): String = {
+ if (tableName.endsWith(OFFLINE_TABLE_SUFFIX)) {
Review comment:
This method is already implemented:
`TableNameBuilder.extractRawTableName()`
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mangrrua edited a comment on pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
mangrrua edited a comment on pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#issuecomment-667951838
> This is great work! Looking forward to streaming API for read and writeAPI :)
>
> One question maybe related to this, shall we create a PinotRDD?
Thank you :)
Spark DatasourceV2 API supports very flexibility and features. We do not need to create PinotRDD. We are working on the dataframe and dataset api. After read data from pinot, all you have to the is play in spark
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] fx19880617 commented on a change in pull request #5787: [Feature] - Spark Pinot Connector
Posted by GitBox <gi...@apache.org>.
fx19880617 commented on a change in pull request #5787:
URL: https://github.com/apache/incubator-pinot/pull/5787#discussion_r467510358
##########
File path: pinot-connectors/pinot-spark-connector/documentation/read_model.md
##########
@@ -0,0 +1,145 @@
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
+# Read Model
+
+Connector can scan offline, hybrid and realtime tables. `table` parameter have to given like below;
Review comment:
Got it, then please specify this in the user doc, or maybe it's better to add one more option as table type?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org