// Open implements the Executor Open interface.
431751
func (e *ChecksumTableExec) Open(ctx context.Context) error {
	if err := e.baseExecutor.Open(ctx); err != nil {
		return err
	}

	concurrency, err := getChecksumTableConcurrency(e.ctx)
	if err != nil {
		return err
	}

	tasks, err := e.buildTasks()
	if err != nil {
		return err
	}

	taskCh := make(chan *checksumTask, len(tasks))
	resultCh := make(chan *checksumResult, len(tasks))
	for i := 0; i < concurrency; i++ {
		go e.checksumWorker(taskCh, resultCh)
	}

	for _, task := range tasks {
		taskCh <- task
	}
	close(taskCh)

	for i := 0; i < len(tasks); i++ {
		result := <-resultCh
		if result.Error != nil {
			err = result.Error
			logutil.Logger(ctx).Error("checksum failed", zap.Error(err))
			continue
		}
		e.handleResult(result)
	}
	if err != nil {
		return err
	}

	return nil
}
514413
func (e *CleanupIndexExec) Open(ctx context.Context) error {
	if err := e.baseExecutor.Open(ctx); err != nil {
		return err
	}
	e.idxChunk = chunk.New(e.getIdxColTypes(), e.initCap, e.maxChunkSize)
	e.idxValues = make(map[int64][][]types.Datum, e.batchSize)
	e.batchKeys = make([]kv.Key, 0, e.batchSize)
	e.idxValsBufs = make([][]types.Datum, e.batchSize)
	sc := e.ctx.GetSessionVars().StmtCtx
	idxKey, _, err := e.index.GenIndexKey(sc, []types.Datum{{}}, math.MinInt64, nil)
	if err != nil {
		return err
	}
	e.lastIdxKey = idxKey
	return nil
}
1190982
func Open(driverName, dataSourceName string) (*DB, error) {
	db, err := sql.Open(driverName, dataSourceName)
	if err != nil {
		return nil, err
	}
	return &DB{
		DB:           db,
		Mapper:       NewCacheMapper(&SnakeMapper{}),
		reflectCache: make(map[reflect.Type]*cacheStruct),
	}, nil
}
989616
func (tpc *TwoPC) Open(dbconfigs *dbconfigs.DBConfigs) {
	tpc.readPool.Open(dbconfigs.AppWithDB(), dbconfigs.DbaWithDB(), dbconfigs.DbaWithDB())
}
631488
func (qe *QueryEngine) Open() error {
	qe.conns.Open(qe.dbconfigs.AppWithDB(), qe.dbconfigs.DbaWithDB(), qe.dbconfigs.AppDebugWithDB())

	conn, err := qe.conns.Get(tabletenv.LocalContext())
	if err != nil {
		qe.conns.Close()
		return err
	}
	qe.binlogFormat, err = conn.VerifyMode(qe.strictTransTables)
	conn.Recycle()

	if err != nil {
		qe.conns.Close()
		return err
	}

	qe.streamConns.Open(qe.dbconfigs.AppWithDB(), qe.dbconfigs.DbaWithDB(), qe.dbconfigs.AppDebugWithDB())
	qe.se.RegisterNotifier("qe", qe.schemaChanged)
	return nil
}
1066905
func (w *Writer) Open() {
	if !w.enabled {
		return
	}
	w.mu.Lock()
	defer w.mu.Unlock()
	if w.isOpen {
		return
	}
	log.Info("Beginning heartbeat writes")
	w.pool.Open(w.dbconfigs.AppWithDB(), w.dbconfigs.DbaWithDB(), w.dbconfigs.AppDebugWithDB())
	w.ticks.Start(func() { w.writeHeartbeat() })
	w.isOpen = true
}
630944
func (s *TranslateFile) Open() (err error) {
	// Open writer & buffered writer.
	if err := os.MkdirAll(filepath.Dir(s.Path), 0777); err != nil {
		return errors.Wrapf(err, "mkdir %s", filepath.Dir(s.Path))
	} else if s.file, err = os.OpenFile(s.Path, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0666); err != nil {
		return errors.Wrapf(err, "open file %s", s.Path)
	}
	s.w = bufio.NewWriter(s.file)

	// Memory map data file.
	if s.data, err = syscall.Mmap(int(s.file.Fd()), 0, s.mapSize, syscall.PROT_READ, syscall.MAP_SHARED); err != nil {
		return errors.Wrapf(err, "creating Mmap (size: %d)", s.mapSize)
	}

	// Replay the log.
	if err := s.replayEntries(); err != nil {
		return errors.Wrap(err, "replaying log entries")
	}

	// Listen to primaryStoreEvents channel.
	s.wg.Add(1)
	go func() { defer s.wg.Done(); s.monitorPrimaryStoreEvents() }()

	return nil
}
627176
func Open(pidfile string) (*os.Process, error) {
	pid, err := readPidFile(pidfile)
	if err != nil {
		return nil, fmt.Errorf("open pid file %s falied, %s", pidfile, err)
	}
	return os.FindProcess(pid)
}
670997
func (b *Broker) open() redis.Conn {
	b.redisOnce.Do(func() {
		b.pool = b.NewPool(b.socketPath, b.host, b.password, b.db, b.GetConfig().Redis, b.GetConfig().TLSConfig)
		b.redsync = redsync.New([]redsync.Pool{b.pool})
	})

	return b.pool.Get()
}
1207047
func Open(driver, dsn string, log EventReceiver) (*Connection, error) {
	if log == nil {
		log = nullReceiver
	}
	conn, err := sql.Open(driver, dsn)
	if err != nil {
		return nil, err
	}
	var d Dialect
	switch driver {
	case "mysql":
		d = dialect.MySQL
	case "postgres":
		d = dialect.PostgreSQL
	case "sqlite3":
		d = dialect.SQLite3
	default:
		return nil, ErrNotSupported
	}
	return &Connection{DB: conn, EventReceiver: log, Dialect: d}, nil
}
316830
func Open(conn base.StreamConnector, path string, cfg interface{}) (base.Stream, error) {
	attrs, err := query.Values(cfg)
	if err != nil {
		return nil, errors.Annotate(err, "failed to generate URL query from config")
	}
	stream, err := conn.ConnectStream(path, attrs)
	if err != nil {
		return nil, errors.Annotatef(err, "cannot connect to %s", path)
	}
	return stream, nil
}
89379
func (s *Server) Open(l int) *Client {
	c := NewClient(l)
	c.Attach(s)
	return c
}
9850
func (w *Writer) open(filepath string) error {
	f, err := os.Create(filepath)
	if err != nil {
		return err
	}
	w.File = f
	return nil
}
886505
func (l *httpFileSystemLoader) Open(name string) (io.ReadCloser, error) {
	if l.fs == nil {
		return nil, &os.PathError{Op: "open", Path: name, Err: os.ErrNotExist}
	}
	return l.fs.Open(name)
}
248992
func Open(listener func(*vecty.Event)) *vecty.EventListener {
	return &vecty.EventListener{Name: "open", Listener: listener}
}
1281868
func (s *Store) Open(ctx context.Context, key string, fetcher Fetcher) (file *File, err error) {
	span, ctx := opentracing.StartSpanFromContext(ctx, "Cached Fetch")
	if s.Component != "" {
		ext.Component.Set(span, s.Component)
	}
	defer func() {
		if err != nil {
			ext.Error.Set(span, true)
			span.SetTag("err", err.Error())
		}
		if file != nil {
			// Update modified time. Modified time is used to decide which
			// files to evict from the cache.
			touch(file.Path)
		}
		span.Finish()
	}()

	if s.Dir == "" {
		return nil, errors.New("diskcache.Store.Dir must be set")
	}

	// path uses a sha256 hash of the key since we want to use it for the
	// disk name.
	h := sha256.Sum256([]byte(key))
	path := filepath.Join(s.Dir, hex.EncodeToString(h[:])) + ".zip"
	span.LogKV("key", key, "path", path)

	// First do a fast-path, assume already on disk
	f, err := os.Open(path)
	if err == nil {
		span.SetTag("source", "fast")
		return &File{File: f, Path: path}, nil
	}

	// We (probably) have to fetch
	span.SetTag("source", "fetch")

	// Do the fetch in another goroutine so we can respect ctx cancellation.
	type result struct {
		f   *File
		err error
	}
	ch := make(chan result, 1)
	go func(ctx context.Context) {
		if s.BackgroundTimeout != 0 {
			var cancel context.CancelFunc
			ctx, cancel = context.WithTimeout(context.Background(), s.BackgroundTimeout)
			defer cancel()
		}
		f, err := doFetch(ctx, path, fetcher)
		ch <- result{f, err}
	}(ctx)

	select {
	case <-ctx.Done():
		// *os.File sets a finalizer to close the file when no longer used, so
		// we don't need to worry about closing the file in the case of context
		// cancellation.
		return nil, ctx.Err()
	case r := <-ch:
		return r.f, r.err
	}
}
204951
func (e *IndexReaderExecutor) Open(ctx context.Context) error {
	var err error
	if e.corColInAccess {
		e.ranges, err = rebuildIndexRanges(e.ctx, e.plans[0].(*plannercore.PhysicalIndexScan), e.idxCols, e.colLens)
		if err != nil {
			return err
		}
	}
	kvRanges, err := distsql.IndexRangesToKVRanges(e.ctx.GetSessionVars().StmtCtx, e.physicalTableID, e.index.ID, e.ranges, e.feedback)
	if err != nil {
		e.feedback.Invalidate()
		return err
	}
	return e.open(ctx, kvRanges)
}