Impl stmt

connection.go

@@ -8,33 +8,36 @@ import (
 	"sqlflow.org/gohive/hiveserver2"
 )
 
-// Options for opened Hive sessions.
-type Options struct {
+// hiveOptions for opened Hive sessions.
+type hiveOptions struct {
 	PollIntervalSeconds int64
 	BatchSize           int64
 }
 
-type Connection struct {
+type hiveConnection struct {
 	thrift  *hiveserver2.TCLIServiceClient
 	session *hiveserver2.TSessionHandle
-	options Options
+	options hiveOptions
 }
 
-func (c *Connection) Begin() (driver.Tx, error) {
+func (c *hiveConnection) Begin() (driver.Tx, error) {
 	return nil, nil
 }
 
-func (c *Connection) Prepare(query string) (driver.Stmt, error) {
-	return nil, nil
+func (c *hiveConnection) Prepare(qry string) (driver.Stmt, error) {
+	if !c.isOpen() {
+		return nil, fmt.Errorf("driver: bad connection")
+	}
+	return &hiveStmt{hc: c, query: qry}, nil
 }
 
-func (c *Connection) isOpen() bool {
+func (c *hiveConnection) isOpen() bool {
 	return c.session != nil
 }
 
 // As hiveserver2 thrift api does not provide Ping method,
 // we use GetInfo instead to check the health of hiveserver2.
-func (c *Connection) Ping(ctx context.Context) (err error) {
+func (c *hiveConnection) Ping(ctx context.Context) (err error) {
 	getInfoReq := hiveserver2.NewTGetInfoReq()
 	getInfoReq.SessionHandle = c.session
 	getInfoReq.InfoType = hiveserver2.TGetInfoType_CLI_SERVER_NAME
@@ -52,7 +55,7 @@ func (c *Connection) Ping(ctx context.Context) (err error) {
 	return nil
 }
 
-func (c *Connection) Close() error {
+func (c *hiveConnection) Close() error {
 	if c.isOpen() {
 		closeReq := hiveserver2.NewTCloseSessionReq()
 		closeReq.SessionHandle = c.session
@@ -74,7 +77,7 @@ func removeLastSemicolon(s string) string {
 	return s
 }
 
-func (c *Connection) execute(ctx context.Context, query string, args []driver.NamedValue) (*hiveserver2.TExecuteStatementResp, error) {
+func (c *hiveConnection) execute(ctx context.Context, query string, args []driver.NamedValue) (*hiveserver2.TExecuteStatementResp, error) {
 	executeReq := hiveserver2.NewTExecuteStatementReq()
 	executeReq.SessionHandle = c.session
 	executeReq.Statement = removeLastSemicolon(query)
@@ -90,15 +93,15 @@ func (c *Connection) execute(ctx context.Context, query string, args []driver.Na
 	return resp, nil
 }
 
-func (c *Connection) QueryContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Rows, error) {
+func (c *hiveConnection) QueryContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Rows, error) {
 	resp, err := c.execute(ctx, query, args)
 	if err != nil {
 		return nil, err
 	}
 	return newRows(c.thrift, resp.OperationHandle, c.options), nil
 }
 
-func (c *Connection) ExecContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Result, error) {
+func (c *hiveConnection) ExecContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Result, error) {
 	resp, err := c.execute(ctx, query, args)
 	if err != nil {
 		return nil, err

driver.go

@@ -49,8 +49,8 @@ func (d drv) Open(dsn string) (driver.Conn, error) {
 		return nil, err
 	}
 
-	options := Options{PollIntervalSeconds: 5, BatchSize: 100000}
-	conn := &Connection{client, session.SessionHandle, options}
+	options := hiveOptions{PollIntervalSeconds: 5, BatchSize: 100000}
+	conn := &hiveConnection{client, session.SessionHandle, options}
 	return conn, nil
 }
 

rows.go

@@ -15,7 +15,7 @@ import (
 type rowSet struct {
 	thrift    *hiveserver2.TCLIServiceClient
 	operation *hiveserver2.TOperationHandle
-	options   Options
+	options   hiveOptions
 
 	columns    []*hiveserver2.TColumnDesc
 	columnStrs []string
@@ -25,10 +25,10 @@ type rowSet struct {
 
 	// resultSet is column-oriented storage format
 	resultSet [][]interface{}
-	status    *Status
+	status    *hiveStatus
 }
 
-type Status struct {
+type hiveStatus struct {
 	state *hiveserver2.TOperationState
 }
 
@@ -151,7 +151,7 @@ func (r *rowSet) poll() error {
 	if resp.OperationState == nil {
 		return errors.New("No error from GetStatus, but nil status!")
 	}
-	r.status = &Status{resp.OperationState}
+	r.status = &hiveStatus{resp.OperationState}
 	return nil
 }
 
@@ -235,7 +235,7 @@ func convertColumn(col *hiveserver2.TColumn) (colValues interface{}, length int)
 	}
 }
 
-func (s Status) isStopped() bool {
+func (s hiveStatus) isStopped() bool {
 	if s.state == nil {
 		return false
 	}
@@ -249,13 +249,13 @@ func (s Status) isStopped() bool {
 	return false
 }
 
-func (s Status) isFinished() bool {
+func (s hiveStatus) isFinished() bool {
 	return s.state != nil && *s.state == hiveserver2.TOperationState_FINISHED_STATE
 }
 
 func newRows(thrift *hiveserver2.TCLIServiceClient,
 	operation *hiveserver2.TOperationHandle,
-	options Options) driver.Rows {
+	options hiveOptions) driver.Rows {
 	return &rowSet{thrift, operation, options, nil, nil,
 		0, nil, nil, nil}
 }

statement.go

@@ -0,0 +1,27 @@
+package gohive
+
+import (
+	"database/sql/driver"
+)
+
+type hiveStmt struct {
+	hc    *hiveConnection
+	query string
+}
+
+func (stmt *hiveStmt) Close() error {
+	panic("not implemented")
+}
+
+func (stmt *hiveStmt) NumInput() int {
+	panic("not implemented")
+}
+
+// Exec accepts stmt like: "INSERT INTO `TABLE` (f1, f2) VALUES(1.3, false)"
+func (stmt *hiveStmt) Exec(args []driver.Value) (driver.Result, error) {
+	panic("not implemented")
+}
+
+func (stmt *hiveStmt) Query(args []driver.Value) (driver.Rows, error) {
+	panic("not implemented")
+}