import org.apache.pinot.client.Connection;
import org.apache.pinot.client.ConnectionFactory;
import org.apache.pinot.client.Request;
import org.apache.pinot.client.ResultSetGroup;
import org.apache.pinot.client.ResultSet;

/**
 * Demonstrates the use of the pinot-client to query Pinot from Java
 */
public class PinotClientExample {

  public static void main(String[] args) {

    // pinot connection
    String zkUrl = "localhost:2181";
    String pinotClusterName = "PinotCluster";
    Connection pinotConnection = ConnectionFactory.fromZookeeper(zkUrl + "/" + pinotClusterName);

    String query = "SELECT COUNT(*) FROM myTable GROUP BY foo";

    // set queryType=sql for querying the sql endpoint
    Request pinotClientRequest = new Request("sql", query);
    ResultSetGroup pinotResultSetGroup = pinotConnection.execute(pinotClientRequest);
    ResultSet resultTableResultSet = pinotResultSetGroup.getResultSet(0);

    int numRows = resultTableResultSet.getRowCount();
    int numColumns = resultTableResultSet.getColumnCount();
    String columnValue = resultTableResultSet.getString(0, 1);
    String columnName = resultTableResultSet.getColumnName(1);

    System.out.println("ColumnName: " + columnName + ", ColumnValue: " + columnValue);
  }
}

ResultSetGroup resultSetGroup = 
  connection.execute(new Request("sql", "select * from foo..."));
// OR
Future<ResultSetGroup> futureResultSetGroup = 
  connection.executeAsync(new Request("sql", "select * from foo..."));

PreparedStatement statement = 
    connection.prepareStatement(new Request("sql", "select * from foo where a = ?"));
statement.setString(1, "bar");

ResultSetGroup resultSetGroup = statement.execute();
// OR
Future<ResultSetGroup> futureResultSetGroup = statement.executeAsync();

Request request = new Request("sql", "select foo, bar from baz where quux = 'quuux'");
ResultSetGroup resultSetGroup = connection.execute(request);
ResultSet resultTableResultSet = pinotResultSetGroup.getResultSet(0);

for (int i = 0; i < resultSet.getRowCount(); ++i) {
  System.out.println("foo: " + resultSet.getString(i, 0));
  System.out.println("bar: " + resultSet.getInt(i, 1));
}

ResultSetGroup resultSetGroup = 
    connection.execute(new Request("pql", "select max(foo), min(foo) from bar"));

System.out.println("Number of result groups:" +
    resultSetGroup.getResultSetCount(); // 2, min(foo) and max(foo)
ResultSet resultSetMax = resultSetGroup.getResultSet(0);
System.out.println("Max foo: " + resultSetMax.getInt(0));
ResultSet resultSetMin = resultSetGroup.getResultSet(1);
System.out.println("Min foo: " + resultSetMin.getInt(0));

ResultSetGroup resultSetGroup = 
    connection.execute(
        new Request("pql", "select min(foo), max(foo) from bar group by baz"));

System.out.println("Number of result groups:" +
    resultSetGroup.getResultSetCount(); // 2, min(foo) and max(foo)

ResultSet minResultSet = resultSetGroup.getResultSet(0);
for(int i = 0; i < minResultSet.length(); ++i) {
    System.out.println("Minimum foo for " + minResultSet.getGroupKeyString(i, 1) +
        ": " + minResultSet.getInt(i));
}

ResultSet maxResultSet = resultSetGroup.getResultSet(1);
for(int i = 0; i < maxResultSet.length(); ++i) {
    System.out.println("Maximum foo for " + maxResultSet.getGroupKeyString(i, 1) +
        ": " + maxResultSet.getInt(i));
}

bin/quick-start-batch.sh

git clone [email protected]:fx19880617/pinot-client-go.git
cd pinot-client-go

go build ./examples/batch-quickstart
./batch-quickstart

pinotClient := pinot.NewFromZookeeper([]string{"localhost:2123"}, "", "QuickStartCluster")

pinotClient := pinot.NewFromBrokerList([]string{"localhost:8000"})

pinotClient := pinot.NewWithConfig(&pinot.ClientConfig{
    ZkConfig: &pinot.ZookeeperConfig{
        ZookeeperPath:     zkPath,
        PathPrefix:        strings.Join([]string{zkPathPrefix, pinotCluster}, "/"),
        SessionTimeoutSec: defaultZkSessionTimeoutSec,
    },
    ExtraHTTPHeader: map[string]string{
        "extra-header":"value",
    },
})

pinotClient, err := pinot.NewFromZookeeper([]string{"localhost:2123"}, "", "QuickStartCluster")
if err != nil {
    log.Error(err)
}
brokerResp, err := pinotClient.ExecuteSQL("baseballStats", "select count(*) as cnt, sum(homeRuns) as sum_homeRuns from baseballStats group by teamID limit 10")
if err != nil {
    log.Error(err)
}
log.Infof("Query Stats: response time - %d ms, scanned docs - %d, total docs - %d", brokerResp.TimeUsedMs, brokerResp.NumDocsScanned, brokerResp.TotalDocs)

type BrokerResponse struct {
    AggregationResults          []*AggregationResult `json:"aggregationResults,omitempty"`
    SelectionResults            *SelectionResults    `json:"SelectionResults,omitempty"`
    ResultTable                 *ResultTable         `json:"resultTable,omitempty"`
    Exceptions                  []Exception          `json:"exceptions"`
    TraceInfo                   map[string]string    `json:"traceInfo,omitempty"`
    NumServersQueried           int                  `json:"numServersQueried"`
    NumServersResponded         int                  `json:"numServersResponded"`
    NumSegmentsQueried          int                  `json:"numSegmentsQueried"`
    NumSegmentsProcessed        int                  `json:"numSegmentsProcessed"`
    NumSegmentsMatched          int                  `json:"numSegmentsMatched"`
    NumConsumingSegmentsQueried int                  `json:"numConsumingSegmentsQueried"`
    NumDocsScanned              int64                `json:"numDocsScanned"`
    NumEntriesScannedInFilter   int64                `json:"numEntriesScannedInFilter"`
    NumEntriesScannedPostFilter int64                `json:"numEntriesScannedPostFilter"`
    NumGroupsLimitReached       bool                 `json:"numGroupsLimitReached"`
    TotalDocs                   int64                `json:"totalDocs"`
    TimeUsedMs                  int                  `json:"timeUsedMs"`
    MinConsumingFreshnessTimeMs int64                `json:"minConsumingFreshnessTimeMs"`
}

// ResultTable is a ResultTable
type ResultTable struct {
    DataSchema RespSchema      `json:"dataSchema"`
    Rows       [][]interface{} `json:"rows"`
}

// RespSchema is response schema
type RespSchema struct {
    ColumnDataTypes []string `json:"columnDataTypes"`
    ColumnNames     []string `json:"columnNames"`
}

func (r ResultTable) GetRowCount() int
func (r ResultTable) GetColumnCount() int
func (r ResultTable) GetColumnName(columnIndex int) string
func (r ResultTable) GetColumnDataType(columnIndex int) string
func (r ResultTable) Get(rowIndex int, columnIndex int) interface{}
func (r ResultTable) GetString(rowIndex int, columnIndex int) string
func (r ResultTable) GetInt(rowIndex int, columnIndex int) int
func (r ResultTable) GetLong(rowIndex int, columnIndex int) int64
func (r ResultTable) GetFloat(rowIndex int, columnIndex int) float32
func (r ResultTable) GetDouble(rowIndex int, columnIndex int) float64

// Print Response Schema
for c := 0; c < brokerResp.ResultTable.GetColumnCount(); c++ {
  fmt.Printf("%s(%s)\t", brokerResp.ResultTable.GetColumnName(c), brokerResp.ResultTable.GetColumnDataType(c))
}
fmt.Println()

// Print Row Table
for r := 0; r < brokerResp.ResultTable.GetRowCount(); r++ {
  for c := 0; c < brokerResp.ResultTable.GetColumnCount(); c++ {
    fmt.Printf("%v\t", brokerResp.ResultTable.Get(r, c))
  }
  fmt.Println()
}